Stream allocator - v0.2 (#216)

* Use protocol friendly StreamedTracksUpdate

* WIP commit

* Stream allocator update

* subtract the requested bandwidth as delta from Allocate could be adding to bandwidth

* Calculate delta correctly

* correct comment

* Simplify eventCh per David's suggestion

pkg/rtc/participant.go

@@ -1314,30 +1314,26 @@ func (p *ParticipantImpl) configureReceiverDTX() {
 	}
 }
 
-func (p *ParticipantImpl) onStreamedTracksChange(paused map[string][]string, resumed map[string][]string) error {
-	if len(paused) == 0 && len(resumed) == 0 {
+func (p *ParticipantImpl) onStreamedTracksChange(update *sfu.StreamedTracksUpdate) error {
+	if len(update.Paused) == 0 && len(update.Resumed) == 0 {
 		return nil
 	}
 
 	streamedTracksUpdate := &livekit.StreamedTracksUpdate{}
-	if len(paused) != 0 {
-		for participantId, trackIds := range paused {
-			for _, trackId := range trackIds {
-				streamedTracksUpdate.Paused = append(streamedTracksUpdate.Paused, &livekit.StreamedTrack{
-					ParticipantSid: participantId,
-					TrackSid:       trackId,
-				})
-			}
+	if len(update.Paused) != 0 {
+		for _, streamedTrack := range update.Paused {
+			streamedTracksUpdate.Paused = append(streamedTracksUpdate.Paused, &livekit.StreamedTrack{
+				ParticipantSid: streamedTrack.ParticipantSid,
+				TrackSid:       streamedTrack.TrackSid,
+			})
 		}
 	}
-	if len(resumed) != 0 {
-		for participantId, trackIds := range paused {
-			for _, trackId := range trackIds {
-				streamedTracksUpdate.Resumed = append(streamedTracksUpdate.Resumed, &livekit.StreamedTrack{
-					ParticipantSid: participantId,
-					TrackSid:       trackId,
-				})
-			}
+	if len(update.Resumed) != 0 {
+		for _, streamedTrack := range update.Resumed {
+			streamedTracksUpdate.Resumed = append(streamedTracksUpdate.Resumed, &livekit.StreamedTrack{
+				ParticipantSid: streamedTrack.ParticipantSid,
+				TrackSid:       streamedTrack.TrackSid,
+			})
 		}
 	}
 

pkg/rtc/transport.go

@@ -256,7 +256,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error {
 	return nil
 }
 
-func (t *PCTransport) OnStreamedTracksChange(f func(paused map[string][]string, resumed map[string][]string) error) {
+func (t *PCTransport) OnStreamedTracksChange(f func(update *sfu.StreamedTracksUpdate) error) {
 	if t.streamAllocator == nil {
 		return
 	}
@@ -269,7 +269,7 @@ func (t *PCTransport) AddTrack(subTrack types.SubscribedTrack) {
 		return
 	}
 
-	t.streamAllocator.AddTrack(subTrack.DownTrack(), subTrack.PublisherIdentity())
+	t.streamAllocator.AddTrack(subTrack.DownTrack())
 }
 
 func (t *PCTransport) RemoveTrack(subTrack types.SubscribedTrack) {

pkg/sfu/buffer/buffer.go

@@ -52,7 +52,7 @@ type Buffer struct {
 	closeOnce  sync.Once
 	mediaSSRC  uint32
 	clockRate  uint32
-	maxBitrate uint64
+	maxBitrate int64
 	lastReport int64
 	twccExt    uint8
 	audioExt   uint8
@@ -69,7 +69,7 @@ type Buffer struct {
 	minPacketProbe     int
 	lastPacketRead     int
 	bitrate            atomic.Value
-	bitrateHelper      [4]uint64
+	bitrateHelper      [4]int64
 	lastSRNTPTime      uint64
 	lastSRRTPTime      uint32
 	lastSRRecv         int64 // Represents wall clock of the most recent sender report arrival
@@ -117,7 +117,7 @@ func NewBuffer(ssrc uint32, vp, ap *sync.Pool, logger logr.Logger) *Buffer {
 		audioPool: ap,
 		logger:    logger,
 	}
-	b.bitrate.Store(make([]uint64, len(b.bitrateHelper)))
+	b.bitrate.Store(make([]int64, len(b.bitrateHelper)))
 	b.extPackets.SetMinCapacity(7)
 	return b
 }
@@ -127,7 +127,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 	defer b.Unlock()
 
 	b.clockRate = codec.ClockRate
-	b.maxBitrate = o.MaxBitRate
+	b.maxBitrate = int64(o.MaxBitRate)
 	b.mime = strings.ToLower(codec.MimeType)
 
 	switch {
@@ -398,7 +398,7 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
 		}
 	}
 
-	b.bitrateHelper[temporalLayer] += uint64(len(pkt))
+	b.bitrateHelper[temporalLayer] += int64(len(pkt))
 
 	diff := arrivalTime - b.lastReport
 	if diff >= ReportDelta {
@@ -408,12 +408,12 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
 		// 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().([]uint64)
+		bitrates, ok := b.bitrate.Load().([]int64)
 		if !ok {
-			bitrates = make([]uint64, len(b.bitrateHelper))
+			bitrates = make([]int64, len(b.bitrateHelper))
 		}
 		for i := 0; i < len(b.bitrateHelper); i++ {
-			br := (8 * b.bitrateHelper[i] * uint64(ReportDelta)) / uint64(diff)
+			br := (8 * b.bitrateHelper[i] * int64(ReportDelta)) / int64(diff)
 			bitrates[i] = br
 			b.bitrateHelper[i] = 0
 		}
@@ -446,10 +446,10 @@ func (b *Buffer) buildNACKPacket() []rtcp.Packet {
 func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate {
 	br := b.Bitrate()
 	if b.stats.LostRate < 0.02 {
-		br = uint64(float64(br)*1.09) + 2000
+		br = int64(float64(br)*1.09) + 2000
 	}
 	if b.stats.LostRate > .1 {
-		br = uint64(float64(br) * float64(1-0.5*b.stats.LostRate))
+		br = int64(float64(br) * float64(1-0.5*b.stats.LostRate))
 	}
 	if br > b.maxBitrate {
 		br = b.maxBitrate
@@ -545,9 +545,9 @@ func (b *Buffer) GetPacket(buff []byte, sn uint16) (int, error) {
 }
 
 // Bitrate returns the current publisher stream bitrate.
-func (b *Buffer) Bitrate() uint64 {
-	bitrates, ok := b.bitrate.Load().([]uint64)
-	bitrate := uint64(0)
+func (b *Buffer) Bitrate() int64 {
+	bitrates, ok := b.bitrate.Load().([]int64)
+	bitrate := int64(0)
 	if ok {
 		for _, b := range bitrates {
 			bitrate += b
@@ -557,14 +557,14 @@ func (b *Buffer) Bitrate() uint64 {
 }
 
 // BitrateTemporalCumulative returns the current publisher stream bitrate temporal layer accumulated with lower temporal layers.
-func (b *Buffer) BitrateTemporalCumulative() []uint64 {
-	bitrates, ok := b.bitrate.Load().([]uint64)
+func (b *Buffer) BitrateTemporalCumulative() []int64 {
+	bitrates, ok := b.bitrate.Load().([]int64)
 	if !ok {
-		return make([]uint64, len(b.bitrateHelper))
+		return make([]int64, len(b.bitrateHelper))
 	}
 
 	// copy and process
-	brs := make([]uint64, len(bitrates))
+	brs := make([]int64, len(bitrates))
 	copy(brs, bitrates)
 
 	for i := len(brs) - 1; i >= 1; i-- {

pkg/sfu/downtrack.go

@@ -22,7 +22,7 @@ import (
 
 // TrackSender defines a  interface send media to remote peer
 type TrackSender interface {
-	UptrackLayersChange(availableLayers []uint16, layerAdded bool)
+	UptrackLayersChange(availableLayers []uint16)
 	WriteRTP(p *buffer.ExtPacket, layer int32) error
 	Close()
 	// ID is the globally unique identifier for this Track.
@@ -110,6 +110,11 @@ type SnTs struct {
 	timestamp      uint32
 }
 
+type VideoLayers struct {
+	spatial  int32
+	temporal int32
+}
+
 type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport)
 
 // DownTrack  implements TrackLocal, is the track used to write packets
@@ -158,13 +163,13 @@ type DownTrack struct {
 	onREMB func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate)
 
 	// simulcast layer availability change callback
-	onAvailableLayersChanged func(dt *DownTrack, layerAdded bool)
+	onAvailableLayersChanged func(dt *DownTrack)
 
 	// subscription change callback
 	onSubscriptionChanged func(dt *DownTrack)
 
 	// max layer change callback
-	onSubscribedLayersChanged func(dt *DownTrack, maxSpatialLayer int32, maxTemporalLayer int32)
+	onSubscribedLayersChanged func(dt *DownTrack, layers VideoLayers)
 
 	// packet sent callback
 	onPacketSent []func(dt *DownTrack, size int)
@@ -467,28 +472,28 @@ func (d *DownTrack) Close() {
 }
 
 func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) {
-	changed := d.forwarder.SetMaxSpatialLayer(spatialLayer)
+	changed, maxLayers := d.forwarder.SetMaxSpatialLayer(spatialLayer)
 	if !changed {
 		return
 	}
 
-	if !d.forwarder.Muted() && d.onSubscribedLayersChanged != nil {
-		d.onSubscribedLayersChanged(d, spatialLayer, d.forwarder.MaxTemporalLayer())
+	if d.onSubscribedLayersChanged != nil {
+		d.onSubscribedLayersChanged(d, maxLayers)
 	}
 }
 
 func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32) {
-	changed := d.forwarder.SetMaxTemporalLayer(temporalLayer)
+	changed, maxLayers := d.forwarder.SetMaxTemporalLayer(temporalLayer)
 	if !changed {
 		return
 	}
 
-	if !d.forwarder.Muted() && d.onSubscribedLayersChanged != nil {
-		d.onSubscribedLayersChanged(d, d.forwarder.MaxSpatialLayer(), temporalLayer)
+	if d.onSubscribedLayersChanged != nil {
+		d.onSubscribedLayersChanged(d, maxLayers)
 	}
 }
 
-func (d *DownTrack) MaxLayers() (int32, int32) {
+func (d *DownTrack) MaxLayers() VideoLayers {
 	return d.forwarder.MaxLayers()
 }
 
@@ -496,9 +501,11 @@ func (d *DownTrack) GetForwardingStatus() ForwardingStatus {
 	return d.forwarder.GetForwardingStatus()
 }
 
-func (d *DownTrack) UptrackLayersChange(availableLayers []uint16, layerAdded bool) {
-	if !d.forwarder.Muted() && d.onAvailableLayersChanged != nil {
-		d.onAvailableLayersChanged(d, layerAdded)
+func (d *DownTrack) UptrackLayersChange(availableLayers []uint16) {
+	d.forwarder.UptrackLayersChange(availableLayers)
+
+	if d.onAvailableLayersChanged != nil {
+		d.onAvailableLayersChanged(d)
 	}
 }
 
@@ -529,7 +536,7 @@ func (d *DownTrack) AddReceiverReportListener(listener ReceiverReportListener) {
 	d.receiverReportListeners = append(d.receiverReportListeners, listener)
 }
 
-func (d *DownTrack) OnAvailableLayersChanged(fn func(dt *DownTrack, layerAdded bool)) {
+func (d *DownTrack) OnAvailableLayersChanged(fn func(dt *DownTrack)) {
 	d.onAvailableLayersChanged = fn
 }
 
@@ -537,7 +544,7 @@ func (d *DownTrack) OnSubscriptionChanged(fn func(dt *DownTrack)) {
 	d.onSubscriptionChanged = fn
 }
 
-func (d *DownTrack) OnSubscribedLayersChanged(fn func(dt *DownTrack, maxSpatialLayer int32, maxTemporalLayer int32)) {
+func (d *DownTrack) OnSubscribedLayersChanged(fn func(dt *DownTrack, layers VideoLayers)) {
 	d.onSubscribedLayersChanged = fn
 }
 
@@ -545,12 +552,28 @@ func (d *DownTrack) OnPacketSent(fn func(dt *DownTrack, size int)) {
 	d.onPacketSent = append(d.onPacketSent, fn)
 }
 
-func (d *DownTrack) AdjustAllocation(availableChannelCapacity uint64) (bool, bool, uint64, uint64) {
-	return d.forwarder.AdjustAllocation(availableChannelCapacity, d.receiver.GetBitrateTemporalCumulative())
+func (d *DownTrack) Allocate(availableChannelCapacity int64) VideoAllocationResult {
+	return d.forwarder.Allocate(availableChannelCapacity, d.receiver.GetBitrateTemporalCumulative())
 }
 
-func (d *DownTrack) IncreaseAllocation() (bool, uint64, uint64) {
-	return d.forwarder.IncreaseAllocation(d.receiver.GetBitrateTemporalCumulative())
+func (d *DownTrack) TryAllocate(additionalChannelCapacity int64) VideoAllocationResult {
+	return d.forwarder.TryAllocate(additionalChannelCapacity, d.receiver.GetBitrateTemporalCumulative())
+}
+
+func (d *DownTrack) FinalizeAllocate() {
+	d.forwarder.FinalizeAllocate(d.receiver.GetBitrateTemporalCumulative())
+}
+
+func (d *DownTrack) AllocateNextHigher() bool {
+	return d.forwarder.AllocateNextHigher(d.receiver.GetBitrateTemporalCumulative())
+}
+
+func (d *DownTrack) AllocationState() VideoAllocationState {
+	return d.forwarder.AllocationState()
+}
+
+func (d *DownTrack) AllocationBandwidth() int64 {
+	return d.forwarder.AllocationBandwidth()
 }
 
 func (d *DownTrack) CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChunk {
@@ -930,6 +953,32 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
 //
 // Forwarder
 //
+type VideoStreamingChange int
+
+const (
+	VideoStreamingChangeNone VideoStreamingChange = iota
+	VideoStreamingChangePausing
+	VideoStreamingChangeResuming
+)
+
+type VideoAllocationState int
+
+const (
+	VideoAllocationStateNone VideoAllocationState = iota
+	VideoAllocationStateMuted
+	VideoAllocationStateFeedDry
+	VideoAllocationStateAwaitingMeasurement
+	VideoAllocationStateOptimal
+	VideoAllocationStateDeficient
+)
+
+type VideoAllocationResult struct {
+	change             VideoStreamingChange
+	state              VideoAllocationState
+	bandwidthRequested int64
+	bandwidthDelta     int64
+}
+
 type Forwarder struct {
 	lock  sync.RWMutex
 	codec webrtc.RTPCodecCapability
@@ -949,6 +998,11 @@ type Forwarder struct {
 	currentTemporalLayer int32
 	targetTemporalLayer  int32
 
+	lastAllocationState      VideoAllocationState
+	lastAllocationRequestBps int64
+
+	availableLayers []uint16
+
 	rtpMunger *RTPMunger
 	vp8Munger *VP8Munger
 }
@@ -964,6 +1018,8 @@ func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo
 		currentTemporalLayer: InvalidTemporalLayer,
 		targetTemporalLayer:  InvalidTemporalLayer,
 
+		lastAllocationState: VideoAllocationStateNone,
+
 		rtpMunger: NewRTPMunger(),
 	}
 
@@ -1001,23 +1057,20 @@ func (f *Forwarder) Muted() bool {
 	return f.muted
 }
 
-func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) bool {
+func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, VideoLayers) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	if spatialLayer == f.maxSpatialLayer {
-		return false
+		return false, VideoLayers{}
 	}
 
 	f.maxSpatialLayer = spatialLayer
-	return true
-}
 
-func (f *Forwarder) MaxSpatialLayer() int32 {
-	f.lock.RLock()
-	defer f.lock.RUnlock()
-
-	return f.maxSpatialLayer
+	return true, VideoLayers{
+		spatial:  f.maxSpatialLayer,
+		temporal: f.maxTemporalLayer,
+	}
 }
 
 func (f *Forwarder) CurrentSpatialLayer() int32 {
@@ -1034,30 +1087,30 @@ func (f *Forwarder) TargetSpatialLayer() int32 {
 	return f.targetSpatialLayer
 }
 
-func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) bool {
+func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, VideoLayers) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	if temporalLayer == f.maxTemporalLayer {
-		return false
+		return false, VideoLayers{}
 	}
 
 	f.maxTemporalLayer = temporalLayer
-	return true
+
+	return true, VideoLayers{
+		spatial:  f.maxSpatialLayer,
+		temporal: f.maxTemporalLayer,
+	}
 }
 
-func (f *Forwarder) MaxTemporalLayer() int32 {
+func (f *Forwarder) MaxLayers() VideoLayers {
 	f.lock.RLock()
 	defer f.lock.RUnlock()
 
-	return f.maxTemporalLayer
-}
-
-func (f *Forwarder) MaxLayers() (int32, int32) {
-	f.lock.RLock()
-	defer f.lock.RUnlock()
-
-	return f.maxSpatialLayer, f.maxTemporalLayer
+	return VideoLayers{
+		spatial:  f.maxSpatialLayer,
+		temporal: f.maxTemporalLayer,
+	}
 }
 
 func (f *Forwarder) GetForwardingStatus() ForwardingStatus {
@@ -1075,73 +1128,23 @@ func (f *Forwarder) GetForwardingStatus() ForwardingStatus {
 	return ForwardingStatusOptimal
 }
 
-func (f *Forwarder) AdjustAllocation(availableChannelCapacity uint64, brs [3][4]uint64) (isPausing, isResuming bool, bandwidthRequested, optimalBandwidthNeeded uint64) {
+func (f *Forwarder) UptrackLayersChange(availableLayers []uint16) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
-	if f.kind == webrtc.RTPCodecTypeAudio || f.muted {
-		return
-	}
-
-	optimalBandwidthNeeded = uint64(0)
-	// LK-TODO for temporal preference, traverse the bitrates array the other way
-	for i := f.maxSpatialLayer; i >= 0; i-- {
-		for j := f.maxTemporalLayer; j >= 0; j-- {
-			if brs[i][j] == 0 {
-				continue
-			}
-			if optimalBandwidthNeeded == 0 {
-				optimalBandwidthNeeded = brs[i][j]
-			}
-			if brs[i][j] < availableChannelCapacity {
-				isResuming = f.targetSpatialLayer == InvalidSpatialLayer
-				bandwidthRequested = brs[i][j]
-
-				f.targetSpatialLayer = int32(i)
-				f.targetTemporalLayer = int32(j)
-				return
-			}
-		}
-	}
-
-	if optimalBandwidthNeeded != 0 {
-		// no layer fits in the available channel capacity, disable the track
-		isPausing = f.targetSpatialLayer != InvalidSpatialLayer
-
-		f.currentSpatialLayer = InvalidSpatialLayer
-		f.targetSpatialLayer = InvalidSpatialLayer
-
-		f.currentTemporalLayer = InvalidTemporalLayer
-		f.targetTemporalLayer = InvalidTemporalLayer
-	}
-	return
+	f.availableLayers = availableLayers
 }
 
-func (f *Forwarder) IncreaseAllocation(brs [3][4]uint64) (increased bool, bandwidthRequested, optimalBandwidthNeeded uint64) {
-	// LK-TODO-START
-	// This is mainly used in probing to try a slightly higher layer.
-	// But, if down track is not a simulcast track, then the next
-	// available layer (i. e. the only layer of simple track) may boost
-	// things by a lot (it could happen in simulcast jumps too).
-	// May need to take in a layer increase threshold as an argument
-	// (in terms of bps) and increase layer only if the jump is within
-	// that threshold.
-	// LK-TODO-END
-	f.lock.Lock()
-	defer f.lock.Unlock()
+func (f *Forwarder) disable() {
+	f.currentSpatialLayer = InvalidSpatialLayer
+	f.targetSpatialLayer = InvalidSpatialLayer
 
-	if f.kind == webrtc.RTPCodecTypeAudio || f.muted {
-		return
-	}
+	f.currentTemporalLayer = InvalidTemporalLayer
+	f.targetTemporalLayer = InvalidTemporalLayer
+}
 
-	// if targets are still pending, don't increase
-	if f.targetSpatialLayer != InvalidSpatialLayer {
-		if f.targetSpatialLayer != f.currentSpatialLayer || f.targetTemporalLayer != f.currentTemporalLayer {
-			return
-		}
-	}
-
-	// move to the next available layer
+func (f *Forwarder) getOptimalBandwidthNeeded(brs [3][4]int64) int64 {
+	optimalBandwidthNeeded := int64(0)
 	for i := f.maxSpatialLayer; i >= 0; i-- {
 		for j := f.maxTemporalLayer; j >= 0; j-- {
 			if brs[i][j] == 0 {
@@ -1157,11 +1160,217 @@ func (f *Forwarder) IncreaseAllocation(brs [3][4]uint64) (increased bool, bandwi
 			break
 		}
 	}
-	if optimalBandwidthNeeded == 0 {
-		// feed is dry
+
+	return optimalBandwidthNeeded
+}
+
+func (f *Forwarder) allocate(availableChannelCapacity int64, canPause bool, brs [3][4]int64) (result VideoAllocationResult) {
+	// should never get called on audio tracks, just for safety
+	if f.kind == webrtc.RTPCodecTypeAudio {
 		return
 	}
 
+	if f.muted {
+		result.state = VideoAllocationStateMuted
+		result.bandwidthRequested = 0
+		result.bandwidthDelta = result.bandwidthRequested - f.lastAllocationRequestBps
+
+		f.lastAllocationState = result.state
+		f.lastAllocationRequestBps = result.bandwidthRequested
+		return
+	}
+
+	optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs)
+	if optimalBandwidthNeeded == 0 {
+		if len(f.availableLayers) == 0 {
+			// feed is dry
+			result.state = VideoAllocationStateFeedDry
+			result.bandwidthRequested = 0
+			result.bandwidthDelta = result.bandwidthRequested - f.lastAllocationRequestBps
+
+			f.lastAllocationState = result.state
+			f.lastAllocationRequestBps = result.bandwidthRequested
+			return
+		}
+
+		// feed bitrate is not yet calculated
+		result.state = VideoAllocationStateAwaitingMeasurement
+		f.lastAllocationState = result.state
+
+		if availableChannelCapacity == ChannelCapacityInfinity {
+			// channel capacity allows a free pass.
+			// So, resume with the highest layer available <= max subscribed layer
+
+			// if already optimistically started, nothing else to do
+			if f.targetSpatialLayer != InvalidSpatialLayer {
+				return
+			}
+
+			f.targetSpatialLayer = int32(f.availableLayers[len(f.availableLayers)-1])
+			if f.targetSpatialLayer > f.maxSpatialLayer {
+				f.targetSpatialLayer = f.maxSpatialLayer
+			}
+
+			f.targetTemporalLayer = f.maxTemporalLayer
+			if f.targetTemporalLayer == InvalidTemporalLayer {
+				f.targetTemporalLayer = 0
+			}
+
+			result.change = VideoStreamingChangeResuming
+		} else {
+			// if not optimistically started, nothing else to do
+			if f.targetSpatialLayer == InvalidSpatialLayer {
+				return
+			}
+
+			if canPause {
+				// disable it as it is not known how big this stream is
+				// and if it will fit in the available channel capacity
+				result.change = VideoStreamingChangePausing
+				result.state = VideoAllocationStateDeficient
+				result.bandwidthRequested = 0
+				result.bandwidthDelta = result.bandwidthRequested - f.lastAllocationRequestBps
+
+				f.lastAllocationState = result.state
+				f.lastAllocationRequestBps = result.bandwidthRequested
+
+				f.disable()
+			}
+		}
+		return
+	}
+
+	// LK-TODO for temporal preference, traverse the bitrates array the other way
+	for i := f.maxSpatialLayer; i >= 0; i-- {
+		for j := f.maxTemporalLayer; j >= 0; j-- {
+			if brs[i][j] == 0 {
+				continue
+			}
+			if brs[i][j] < availableChannelCapacity {
+				if f.targetSpatialLayer == InvalidSpatialLayer {
+					result.change = VideoStreamingChangeResuming
+				}
+				result.bandwidthRequested = brs[i][j]
+				result.bandwidthDelta = result.bandwidthRequested - f.lastAllocationRequestBps
+				if result.bandwidthRequested == optimalBandwidthNeeded {
+					result.state = VideoAllocationStateOptimal
+				} else {
+					result.state = VideoAllocationStateDeficient
+				}
+
+				f.lastAllocationState = result.state
+				f.lastAllocationRequestBps = result.bandwidthRequested
+
+				f.targetSpatialLayer = int32(i)
+				f.targetTemporalLayer = int32(j)
+				return
+			}
+		}
+	}
+
+	if !canPause {
+		// do not pause if preserving
+		// although preserving, currently streamed layers could have a different bitrate,
+		// but not updating to prevent entropy increase.
+		result.state = f.lastAllocationState
+		return
+	}
+
+	// no layer fits in the available channel capacity, disable the track
+	if f.targetSpatialLayer != InvalidSpatialLayer {
+		result.change = VideoStreamingChangePausing
+	}
+	result.state = VideoAllocationStateDeficient
+	result.bandwidthRequested = 0
+	result.bandwidthDelta = result.bandwidthRequested - f.lastAllocationRequestBps
+
+	f.lastAllocationState = result.state
+	f.lastAllocationRequestBps = result.bandwidthRequested
+
+	f.disable()
+	return
+}
+
+func (f *Forwarder) Allocate(availableChannelCapacity int64, brs [3][4]int64) VideoAllocationResult {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	return f.allocate(availableChannelCapacity, true, brs)
+}
+
+func (f *Forwarder) TryAllocate(additionalChannelCapacity int64, brs [3][4]int64) VideoAllocationResult {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	return f.allocate(f.lastAllocationRequestBps+additionalChannelCapacity, false, brs)
+}
+
+func (f *Forwarder) FinalizeAllocate(brs [3][4]int64) {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	if f.lastAllocationState != VideoAllocationStateAwaitingMeasurement {
+		return
+	}
+
+	optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs)
+	if optimalBandwidthNeeded == 0 {
+		if len(f.availableLayers) == 0 {
+			// feed dry
+			f.lastAllocationState = VideoAllocationStateFeedDry
+			f.lastAllocationRequestBps = 0
+		}
+
+		// still awaiting measurement
+		return
+	}
+
+	// LK-TODO for temporal preference, traverse the bitrates array the other way
+	for i := f.maxSpatialLayer; i >= 0; i-- {
+		for j := f.maxTemporalLayer; j >= 0; j-- {
+			if brs[i][j] == 0 {
+				continue
+			}
+
+			f.lastAllocationState = VideoAllocationStateOptimal
+			f.lastAllocationRequestBps = brs[i][j]
+
+			f.targetSpatialLayer = int32(i)
+			f.targetTemporalLayer = int32(j)
+			break
+		}
+	}
+}
+
+func (f *Forwarder) AllocateNextHigher(brs [3][4]int64) bool {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	if f.kind == webrtc.RTPCodecTypeAudio {
+		return false
+	}
+
+	// if targets are still pending, don't increase
+	if f.targetSpatialLayer != InvalidSpatialLayer {
+		if f.targetSpatialLayer != f.currentSpatialLayer || f.targetTemporalLayer != f.currentTemporalLayer {
+			return false
+		}
+	}
+
+	optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs)
+	if optimalBandwidthNeeded == 0 {
+		if len(f.availableLayers) == 0 {
+			f.lastAllocationState = VideoAllocationStateFeedDry
+			f.lastAllocationRequestBps = 0
+			return false
+		}
+
+		// bitrates not available yet
+		f.lastAllocationState = VideoAllocationStateAwaitingMeasurement
+		f.lastAllocationRequestBps = 0
+		return false
+	}
+
 	// try moving temporal layer up in the current spatial layer
 	nextTemporalLayer := f.currentTemporalLayer + 1
 	currentSpatialLayer := f.currentSpatialLayer
@@ -1172,9 +1381,13 @@ func (f *Forwarder) IncreaseAllocation(brs [3][4]uint64) (increased bool, bandwi
 		f.targetSpatialLayer = currentSpatialLayer
 		f.targetTemporalLayer = nextTemporalLayer
 
-		increased = true
-		bandwidthRequested = brs[currentSpatialLayer][nextTemporalLayer]
-		return
+		f.lastAllocationRequestBps = brs[currentSpatialLayer][nextTemporalLayer]
+		if f.lastAllocationRequestBps < optimalBandwidthNeeded {
+			f.lastAllocationState = VideoAllocationStateDeficient
+		} else {
+			f.lastAllocationState = VideoAllocationStateOptimal
+		}
+		return true
 	}
 
 	// try moving spatial layer up if already at max temporal layer of current spatial layer
@@ -1183,12 +1396,30 @@ func (f *Forwarder) IncreaseAllocation(brs [3][4]uint64) (increased bool, bandwi
 		f.targetSpatialLayer = nextSpatialLayer
 		f.targetTemporalLayer = 0
 
-		increased = true
-		bandwidthRequested = brs[nextSpatialLayer][0]
-		return
+		f.lastAllocationRequestBps = brs[nextSpatialLayer][0]
+		if f.lastAllocationRequestBps < optimalBandwidthNeeded {
+			f.lastAllocationState = VideoAllocationStateDeficient
+		} else {
+			f.lastAllocationState = VideoAllocationStateOptimal
+		}
+		return true
 	}
 
-	return
+	return false
+}
+
+func (f *Forwarder) AllocationState() VideoAllocationState {
+	f.lock.RLock()
+	defer f.lock.RUnlock()
+
+	return f.lastAllocationState
+}
+
+func (f *Forwarder) AllocationBandwidth() int64 {
+	f.lock.RLock()
+	defer f.lock.RUnlock()
+
+	return f.lastAllocationRequestBps
 }
 
 func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32) (*TranslationParams, error) {

pkg/sfu/prober.go

@@ -129,7 +129,7 @@ type Prober struct {
 	clusters      deque.Deque
 	activeCluster *Cluster
 
-	onSendProbe func(bytesToSend int) int
+	onSendProbe func(bytesToSend int)
 }
 
 func NewProber(params ProberParams) *Prober {
@@ -160,7 +160,7 @@ func (p *Prober) Reset() {
 	p.activeCluster = nil
 }
 
-func (p *Prober) OnSendProbe(f func(bytesToSend int) int) {
+func (p *Prober) OnSendProbe(f func(bytesToSend int)) {
 	p.onSendProbe = f
 }
 
@@ -184,6 +184,15 @@ func (p *Prober) PacketSent(size int) {
 	cluster.PacketSent(size)
 }
 
+func (p *Prober) ProbeSent(size int) {
+	cluster := p.getFrontCluster()
+	if cluster == nil {
+		return
+	}
+
+	cluster.ProbeSent(size)
+}
+
 func (p *Prober) getFrontCluster() *Cluster {
 	p.clustersMu.RLock()
 	defer p.clustersMu.RUnlock()
@@ -246,7 +255,9 @@ func (p *Prober) run() {
 			return
 		}
 
-		if !cluster.Process(p) {
+		cluster.Process(p)
+
+		if cluster.IsFinished() {
 			p.logger.Debugw("cluster finished", "participant", p.participantID, "cluster", cluster.String())
 			p.popFrontCluster(cluster)
 			continue
@@ -316,16 +327,37 @@ func (c *Cluster) PacketSent(size int) {
 	c.bytesSentNonProbe += size
 }
 
-func (c *Cluster) Process(p *Prober) bool {
+func (c *Cluster) ProbeSent(size int) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	c.bytesSentProbe += size
+}
+
+func (c *Cluster) IsFinished() bool {
 	c.lock.RLock()
+	defer c.lock.RUnlock()
 
 	// if already past deadline, end the cluster
 	timeElapsed := time.Since(c.startTime)
 	if timeElapsed > c.maxDuration {
-		c.lock.RUnlock()
-		return false
+		return true
 	}
 
+	// do not end cluster until minDuration elapses even if rate is achieved.
+	// Ensures that the next cluster (if any) does not start early.
+	if (c.bytesSentProbe+c.bytesSentNonProbe) >= c.desiredBytes && timeElapsed >= c.minDuration {
+		return true
+	}
+
+	return false
+}
+
+func (c *Cluster) Process(p *Prober) {
+	c.lock.RLock()
+
+	timeElapsed := time.Since(c.startTime)
+
 	// Calculate number of probe bytes that should have been sent since start.
 	// Overall goal is to send desired number of probe bytes in minDuration.
 	// However it is possible that timeElapsed is more than minDuration due
@@ -351,24 +383,11 @@ func (c *Cluster) Process(p *Prober) bool {
 	bytesShortFall = ((bytesShortFall + 274) / 275) * 275
 	c.lock.RUnlock()
 
-	bytesSent := 0
 	if bytesShortFall > 0 && p.onSendProbe != nil {
-		bytesSent = p.onSendProbe(bytesShortFall)
-	}
-
-	c.lock.Lock()
-	c.bytesSentProbe += bytesSent
-
-	// do not end cluster until minDuration elapses even if rate is achieved.
-	// Ensures that the next cluster (if any) does not start early.
-	if (c.bytesSentProbe+c.bytesSentNonProbe) >= c.desiredBytes && timeElapsed >= c.minDuration {
-		c.lock.Unlock()
-		return false
+		p.onSendProbe(bytesShortFall)
 	}
 
 	// LK-TODO look at adapting sleep time based on how many bytes and how much time is left
-	c.lock.Unlock()
-	return true
 }
 
 func (c *Cluster) String() string {

pkg/sfu/receiver.go

@@ -4,6 +4,7 @@ import (
 	"io"
 	"math/rand"
 	"runtime"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -19,7 +20,7 @@ import (
 type TrackReceiver interface {
 	TrackID() string
 	StreamID() string
-	GetBitrateTemporalCumulative() [3][4]uint64
+	GetBitrateTemporalCumulative() [3][4]int64
 	ReadRTP(buf []byte, layer uint8, sn uint16) (int, error)
 	AddDownTrack(track TrackSender)
 	DeleteDownTrack(peerID string)
@@ -37,7 +38,7 @@ type Receiver interface {
 	AddDownTrack(track TrackSender)
 	SetUpTrackPaused(paused bool)
 	NumAvailableSpatialLayers() int
-	GetBitrateTemporalCumulative() [3][4]uint64
+	GetBitrateTemporalCumulative() [3][4]int64
 	ReadRTP(buf []byte, layer uint8, sn uint16) (int, error)
 	DeleteDownTrack(ID string)
 	OnCloseHandler(fn func())
@@ -48,10 +49,6 @@ type Receiver interface {
 	DebugInfo() map[string]interface{}
 }
 
-var (
-	defaultBitratesCumulative = [][]uint64{{60000, 90000, 150000, 0}, {200000, 300000, 500000, 0}, {400000, 600000, 1000000, 0}}
-)
-
 // WebRTCReceiver receives a video track
 type WebRTCReceiver struct {
 	peerID          string
@@ -250,7 +247,7 @@ func (w *WebRTCReceiver) AddDownTrack(track TrackSender) {
 		layers, ok := w.availableLayers.Load().([]uint16)
 		w.upTrackMu.RUnlock()
 		if ok && len(layers) != 0 {
-			track.UptrackLayersChange(layers, true)
+			track.UptrackLayersChange(layers)
 		}
 	}
 
@@ -280,12 +277,12 @@ func (w *WebRTCReceiver) NumAvailableSpatialLayers() int {
 	return len(layers)
 }
 
-func (w *WebRTCReceiver) downtrackLayerChange(layers []uint16, layerAdded bool) {
+func (w *WebRTCReceiver) downtrackLayerChange(layers []uint16) {
 	w.downTrackMu.RLock()
 	defer w.downTrackMu.RUnlock()
 	for _, dt := range w.downTracks {
 		if dt != nil {
-			dt.UptrackLayersChange(layers, layerAdded)
+			dt.UptrackLayersChange(layers)
 		}
 	}
 }
@@ -306,10 +303,11 @@ func (w *WebRTCReceiver) addAvailableLayer(layer uint16) {
 	if !hasLayer {
 		layers = append(layers, layer)
 	}
+	sort.Slice(layers, func(i, j int) bool { return layers[i] < layers[j] })
 	w.availableLayers.Store(layers)
 	w.upTrackMu.Unlock()
 
-	w.downtrackLayerChange(layers, true)
+	w.downtrackLayerChange(layers)
 }
 
 func (w *WebRTCReceiver) removeAvailableLayer(layer uint16) {
@@ -325,23 +323,24 @@ func (w *WebRTCReceiver) removeAvailableLayer(layer uint16) {
 			newLayers = append(newLayers, l)
 		}
 	}
+	sort.Slice(newLayers, func(i, j int) bool { return newLayers[i] < newLayers[j] })
 	w.availableLayers.Store(newLayers)
 	w.upTrackMu.Unlock()
+
 	// need to immediately switch off unavailable layers
-	w.downtrackLayerChange(newLayers, false)
+	w.downtrackLayerChange(newLayers)
 }
 
-func (w *WebRTCReceiver) GetBitrateTemporalCumulative() [3][4]uint64 {
+func (w *WebRTCReceiver) GetBitrateTemporalCumulative() [3][4]int64 {
 	// LK-TODO: For SVC tracks, need to accumulate across spatial layers also
-	var br [3][4]uint64
+	var br [3][4]int64
 	w.bufferMu.RLock()
 	defer w.bufferMu.RUnlock()
 	for i, buff := range w.buffers {
 		if buff != nil {
-			tls := make([]uint64, 4)
+			tls := make([]int64, 4)
 			if w.hasSpatialLayer(int32(i)) {
 				tls = buff.BitrateTemporalCumulative()
-				MaybeUseDefaultBitrate(tls, i)
 			}
 
 			for j := 0; j < len(br[i]); j++ {
@@ -525,32 +524,6 @@ func (w *WebRTCReceiver) storeDownTrack(track TrackSender) {
 	w.downTracks = append(w.downTracks, track)
 }
 
-func MaybeUseDefaultBitrate(tlbs []uint64, layer int) {
-	for _, tlb := range tlbs {
-		if tlb != uint64(0) {
-			// some layer has data
-			return
-		}
-	}
-
-	//
-	// Before measured bitrate is available, initialize with some sane default values.
-	// Note that not all clients send temporal layers or have same number of temporal layers.
-	//   o Safari 15 - does not do temporal layers
-	//   o Chrome 95 - does two temporal layers
-	//   o Firefox 94 - does three temporal layers
-	// Default initialization is to ensure that StreamAllocator has some data and can
-	// forward tracks as soon as available.
-	//
-	// Measured bitrate will be available periodically starting shortly after stream
-	// starts flowing (see `reportDelta` in buffer.go). Once that information becomes
-	// available, it will be used for stream allocation.
-	//
-	for i := 0; i < len(tlbs); i++ {
-		tlbs[i] = defaultBitratesCumulative[layer][i]
-	}
-}
-
 func (w *WebRTCReceiver) DebugInfo() map[string]interface{} {
 	info := map[string]interface{}{
 		"Simulcast": w.isSimulcast,