dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-29 09:49:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify layer roll back. (#2702)

Browse Source 
* Cache TS offset along with sender report.

* set offset after clearing

* simplify roll back
This commit is contained in:
Raja Subramanian
2024-05-04 15:00:19 +05:30
committed by GitHub
parent 45ed030ce6
commit 4d7586bda8
4 changed files with 55 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/forwarder.go
+47 -41
View File
@@ -215,6 +215,11 @@ func (f ForwarderState) String() string {
// -------------------------------------------------------------------
type refInfo struct {
senderReport *buffer.RTCPSenderReportData
tsOffset uint64
}
type Forwarder struct {
lock sync.RWMutex
codec webrtc.RTPCodecCapability
@@ -233,7 +238,7 @@ type Forwarder struct {
lastSSRC uint32
referenceLayerSpatial int32
dummyStartTSOffset uint64
refSenderReports [buffer.DefaultMaxLayerSpatial + 1]*buffer.RTCPSenderReportData
refInfos [buffer.DefaultMaxLayerSpatial + 1]refInfo
refIsSVC bool
provisional *VideoAllocationProvisional
@@ -326,15 +331,17 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [
f.vls = videolayerselector.NewSimulcast(f.logger)
}
f.vls.SetTemporalLayerSelector(temporallayerselector.NewVP8(f.logger))
case "video/h264":
if f.vls != nil {
f.vls = videolayerselector.NewSimulcastFromNull(f.vls)
} else {
f.vls = videolayerselector.NewSimulcast(f.logger)
}
case "video/vp9":
isDDAvailable := ddAvailable(extensions)
case "video/vp9":
// DD-TODO : we only enable dd layer selector for av1/vp9 now, in the future we can enable it for vp8 too
isDDAvailable := ddAvailable(extensions)
if isDDAvailable {
if f.vls != nil {
f.vls = videolayerselector.NewDependencyDescriptorFromNull(f.vls)
@@ -349,9 +356,9 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [
}
}
// SVC-TODO: Support for VP9 simulcast. When DD is not available, have to pick selector based on VP9 SVC or Simulcast
case "video/av1":
// DD-TODO : we only enable dd layer selector for av1/vp9 now, in the future we can enable it for vp8 too
isDDAvailable := ddAvailable(extensions)
if isDDAvailable {
if f.vls != nil {
@@ -568,8 +575,8 @@ func (f *Forwarder) SetRefSenderReport(isSVC bool, layer int32, srData *buffer.R
if isSVC {
layer = 0
}
if layer >= 0 && int(layer) < len(f.refSenderReports) {
f.refSenderReports[layer] = srData
if layer >= 0 && int(layer) < len(f.refInfos) {
f.refInfos[layer].senderReport = srData
}
}
@@ -604,7 +611,7 @@ func (f *Forwarder) clearRefSenderReportsLocked() {
// By clearing sender report on (re)start of a stream, subscribers will wait for a fresh report
// after unmute to send sender report.
for layer := int32(0); layer < buffer.DefaultMaxLayerSpatial+1; layer++ {
f.refSenderReports[layer] = nil
f.refInfos[layer] = refInfo{nil, 0}
}
}
@@ -613,20 +620,22 @@ func (f *Forwarder) GetSenderReportParams() (int32, uint64, *buffer.RTCPSenderRe
defer f.lock.RUnlock()
if f.kind == webrtc.RTPCodecTypeAudio {
return 0, f.rtpMunger.GetPinnedTSOffset(), f.refSenderReports[0]
return 0, f.refInfos[0].tsOffset, f.refInfos[0].senderReport
}
currentLayerSpatial := f.vls.GetCurrent().Spatial
if currentLayerSpatial < 0 || currentLayerSpatial > buffer.DefaultMaxLayerSpatial {
return currentLayerSpatial, f.rtpMunger.GetPinnedTSOffset(), nil
return currentLayerSpatial, 0, nil
}
refSenderReport := f.refSenderReports[currentLayerSpatial]
refSenderReport := f.refInfos[currentLayerSpatial].senderReport
tsOffset := f.refInfos[currentLayerSpatial].tsOffset
if f.refIsSVC {
refSenderReport = f.refSenderReports[0]
refSenderReport = f.refInfos[0].senderReport
tsOffset = f.refInfos[0].tsOffset
}
return currentLayerSpatial, f.rtpMunger.GetPinnedTSOffset(), refSenderReport
return currentLayerSpatial, tsOffset, refSenderReport
}
func (f *Forwarder) isDeficientLocked() bool {
@@ -1534,6 +1543,7 @@ func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32)
switch f.kind {
case webrtc.RTPCodecTypeAudio:
return f.getTranslationParamsAudio(extPkt, layer)
case webrtc.RTPCodecTypeVideo:
return f.getTranslationParamsVideo(extPkt, layer)
}
@@ -1544,7 +1554,7 @@ func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32)
}
func (f *Forwarder) getReferenceLayerRTPTimestamp(ts uint32, refLayer, targetLayer int32) (uint32, error) {
if refLayer < 0 || int(refLayer) > len(f.refSenderReports) || targetLayer < 0 || int(targetLayer) > len(f.refSenderReports) {
if refLayer < 0 || int(refLayer) > len(f.refInfos) || targetLayer < 0 || int(targetLayer) > len(f.refInfos) {
return 0, fmt.Errorf("invalid layer(s), refLayer: %d, targetLayer: %d", refLayer, targetLayer)
}
@@ -1552,8 +1562,8 @@ func (f *Forwarder) getReferenceLayerRTPTimestamp(ts uint32, refLayer, targetLay
return ts, nil
}
srRef := f.refSenderReports[refLayer]
srTarget := f.refSenderReports[targetLayer]
srRef := f.refInfos[refLayer].senderReport
srTarget := f.refInfos[targetLayer].senderReport
if srRef == nil || srRef.NTPTimestamp == 0 || srTarget == nil || srTarget.NTPTimestamp == 0 {
return 0, fmt.Errorf("unavailable layer(s), refLayer: %d, targetLayer: %d", refLayer, targetLayer)
}
@@ -1576,7 +1586,10 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) e
f.referenceLayerSpatial = layer
f.rtpMunger.SetLastSnTs(extPkt)
f.codecMunger.SetLast(extPkt)
f.clearRefSenderReportsLocked()
f.refInfos[layer].tsOffset = f.rtpMunger.GetTSOffset()
f.logger.Debugw(
"starting forwarding",
"sequenceNumber", extPkt.Packet.SequenceNumber,
@@ -1776,16 +1789,18 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) e
)
f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, extNextTS-extLastTS)
f.refInfos[layer].tsOffset = f.rtpMunger.GetTSOffset()
f.codecMunger.UpdateOffsets(extPkt)
return nil
}
// should be called with lock held
func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer int32, tp *TranslationParams) (bool, error) {
func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer int32, tp *TranslationParams) error {
if f.lastSSRC != extPkt.Packet.SSRC {
if err := f.processSourceSwitch(extPkt, layer); err != nil {
tp.shouldDrop = true
return false, nil
f.vls.Rollback()
return nil
}
f.logger.Debugw("switching feed", "from", f.lastSSRC, "to", extPkt.Packet.SSRC)
f.lastSSRC = extPkt.Packet.SSRC
@@ -1795,9 +1810,9 @@ func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer i
if err != nil {
tp.shouldDrop = true
if err == ErrPaddingOnlyPacket || err == ErrDuplicatePacket || err == ErrOutOfOrderSequenceNumberCacheMiss {
return false, nil
return nil
}
return false, err
return err
}
tp.rtp = tpRTP
@@ -1806,13 +1821,13 @@ func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer i
return f.translateCodecHeader(extPkt, tp)
}
return false, nil
return nil
}
// should be called with lock held
func (f *Forwarder) getTranslationParamsAudio(extPkt *buffer.ExtPacket, layer int32) (TranslationParams, error) {
tp := TranslationParams{}
if _, err := f.getTranslationParamsCommon(extPkt, layer, &tp); err != nil {
if err := f.getTranslationParamsCommon(extPkt, layer, &tp); err != nil {
tp.shouldDrop = true
return tp, err
}
@@ -1821,12 +1836,6 @@ func (f *Forwarder) getTranslationParamsAudio(extPkt *buffer.ExtPacket, layer in
// should be called with lock held
func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer int32) (TranslationParams, error) {
maybeRollback := func(isSwitching bool) {
if isSwitching {
f.vls.Rollback()
}
}
tp := TranslationParams{}
if !f.vls.GetTarget().IsValid() {
// stream is paused by streamallocator
@@ -1852,6 +1861,11 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in
tp.ddBytes = result.DependencyDescriptorExtension
tp.marker = result.RTPMarker
err := f.getTranslationParamsCommon(extPkt, layer, &tp)
if tp.shouldDrop {
return tp, err
}
if FlagPauseOnDowngrade && f.isDeficientLocked() && f.vls.GetTarget().Spatial < f.vls.GetCurrent().Spatial {
//
// If target layer is lower than both the current and
@@ -1872,22 +1886,15 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in
// To differentiate between the two cases, drop only when in DEFICIENT state.
//
tp.shouldDrop = true
maybeRollback(result.IsSwitching)
return tp, nil
}
isTemporalSwitching, err := f.getTranslationParamsCommon(extPkt, layer, &tp)
if tp.shouldDrop {
maybeRollback(result.IsSwitching || isTemporalSwitching)
return tp, err
}
return tp, err
return tp, nil
}
func (f *Forwarder) translateCodecHeader(extPkt *buffer.ExtPacket, tp *TranslationParams) (bool, error) {
func (f *Forwarder) translateCodecHeader(extPkt *buffer.ExtPacket, tp *TranslationParams) error {
// codec specific forwarding check and any needed packet munging
tl, isSwitching := f.vls.SelectTemporal(extPkt)
tl := f.vls.SelectTemporal(extPkt)
inputSize, codecBytes, err := f.codecMunger.UpdateAndGet(
extPkt,
tp.rtp.snOrdering == SequenceNumberOrderingOutOfOrder,
@@ -1901,15 +1908,14 @@ func (f *Forwarder) translateCodecHeader(extPkt *buffer.ExtPacket, tp *Translati
// filtered temporal layer, update sequence number offset to prevent holes
f.rtpMunger.PacketDropped(extPkt)
}
return isSwitching, nil
return nil
}
return isSwitching, err
return err
}
tp.incomingHeaderSize = inputSize
tp.codecBytes = codecBytes
return isSwitching, nil
return nil
}
func (f *Forwarder) maybeStart() {
pkg/sfu/rtpmunger.go
+2 -6
View File
@@ -83,7 +83,6 @@ type RTPMunger struct {
extLastTS uint64
extSecondLastTS uint64
tsOffset uint64
pinnedTSOffset uint64
lastMarker bool
secondLastMarker bool
@@ -108,7 +107,6 @@ func (r *RTPMunger) DebugInfo() map[string]interface{} {
"ExtLastTS": r.extLastTS,
"ExtSecondLastTS": r.extSecondLastTS,
"TSOffset": r.tsOffset,
"PinnedTSOffset": r.pinnedTSOffset,
"LastMarker": r.lastMarker,
"SecondLastMarker": r.secondLastMarker,
}
@@ -125,8 +123,8 @@ func (r *RTPMunger) GetLast() RTPMungerState {
}
}
func (r *RTPMunger) GetPinnedTSOffset() uint64 {
return r.pinnedTSOffset
func (r *RTPMunger) GetTSOffset() uint64 {
return r.tsOffset
}
func (r *RTPMunger) SeedLast(state RTPMungerState) {
@@ -149,7 +147,6 @@ func (r *RTPMunger) SetLastSnTs(extPkt *buffer.ExtPacket) {
r.extLastTS = extPkt.ExtTimestamp
r.extSecondLastTS = extPkt.ExtTimestamp
r.tsOffset = 0
r.pinnedTSOffset = r.tsOffset
}
func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint64, tsAdjust uint64) {
@@ -159,7 +156,6 @@ func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint64,
r.updateSnOffset()
r.tsOffset = extPkt.ExtTimestamp - r.extLastTS - tsAdjust
r.pinnedTSOffset = r.tsOffset
}
func (r *RTPMunger) PacketDropped(extPkt *buffer.ExtPacket) {
pkg/sfu/videolayerselector/base.go
+5 -9
View File
@@ -140,29 +140,25 @@ func (b *Base) Rollback() {
b.targetLayer = b.previousTargetLayer
}
func (b *Base) SelectTemporal(extPkt *buffer.ExtPacket) (int32, bool) {
func (b *Base) SelectTemporal(extPkt *buffer.ExtPacket) int32 {
if b.tls != nil {
isSwitching := false
this, next := b.tls.Select(extPkt, b.currentLayer.Temporal, b.targetLayer.Temporal)
if next != b.currentLayer.Temporal {
isSwitching = true
b.previousLayer = b.currentLayer
previousLayer := b.currentLayer
b.currentLayer.Temporal = next
b.logger.Debugw(
"updating temporal layer",
"previous", b.previousLayer,
"previous", previousLayer,
"current", b.currentLayer,
"previousTarget", b.previousTargetLayer,
"target", b.targetLayer,
"max", b.maxLayer,
"req", b.requestSpatial,
"maxSeen", b.maxSeenLayer,
)
}
return this, isSwitching
return this
}
return b.currentLayer.Temporal, false
return b.currentLayer.Temporal
}
pkg/sfu/videolayerselector/videolayerselector.go
+1 -1
View File
@@ -55,6 +55,6 @@ type VideoLayerSelector interface {
GetCurrent() buffer.VideoLayer
Select(extPkt *buffer.ExtPacket, layer int32) VideoLayerSelectorResult
SelectTemporal(extPkt *buffer.ExtPacket) (int32, bool)
SelectTemporal(extPkt *buffer.ExtPacket) int32
Rollback()
}