mirror of
https://github.com/livekit/livekit.git
synced 2026-07-20 22:11:10 +00:00
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
This commit is contained in:
Raja Subramanian
+23
-117
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
+19
-7
@@ -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)
|
||||
|
||||
+72
-218
@@ -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 {
|
||||
|
||||
+19
-222
@@ -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,
|
||||
|
||||
+14
-17
@@ -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()
|
||||
|
||||
+29
-19
@@ -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()
|
||||
}
|
||||
|
||||
+150
-50
@@ -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()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user