diff --git a/pkg/sfu/buffer/dependencydescriptorparser.go b/pkg/sfu/buffer/dependencydescriptorparser.go index 6c91af260..8cc648b7f 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser.go +++ b/pkg/sfu/buffer/dependencydescriptorparser.go @@ -52,8 +52,8 @@ func NewDependencyDescriptorParser(ddExtID uint8, logger logger.Logger, onMaxLay ddExtID: ddExtID, logger: logger, onMaxLayerChanged: onMaxLayerChanged, - seqWrapAround: utils.NewWrapAround[uint16, uint64](), - frameWrapAround: utils.NewWrapAround[uint16, uint64](), + seqWrapAround: utils.NewWrapAround[uint16, uint64](utils.WrapAroundParams{IsRestartAllowed: false}), + frameWrapAround: utils.NewWrapAround[uint16, uint64](utils.WrapAroundParams{IsRestartAllowed: false}), frameChecker: NewFrameIntegrityChecker(180, 1024), // 2seconds for L3T3 30fps video } } diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go index dad7c66ef..b621f42b1 100644 --- a/pkg/sfu/buffer/rtpstats_receiver.go +++ b/pkg/sfu/buffer/rtpstats_receiver.go @@ -60,8 +60,8 @@ type RTPStatsReceiver struct { func NewRTPStatsReceiver(params RTPStatsParams) *RTPStatsReceiver { return &RTPStatsReceiver{ rtpStatsBase: newRTPStatsBase(params), - sequenceNumber: utils.NewWrapAround[uint16, uint64](), - timestamp: utils.NewWrapAround[uint32, uint64](), + sequenceNumber: utils.NewWrapAround[uint16, uint64](utils.WrapAroundParams{IsRestartAllowed: false}), + timestamp: utils.NewWrapAround[uint32, uint64](utils.WrapAroundParams{IsRestartAllowed: false}), history: protoutils.NewBitmap[uint64](cHistorySize), } } @@ -123,36 +123,16 @@ func (r *RTPStatsReceiver) Update( ) } else { resSN = r.sequenceNumber.Update(sequenceNumber) + if resSN.IsUnhandled { + flowState.IsNotHandled = true + return + } resTS = r.timestamp.Update(timestamp) } pktSize := uint64(hdrSize + payloadSize + paddingSize) gapSN := int64(resSN.ExtendedVal - resSN.PreExtendedHighest) if gapSN <= 0 { // duplicate OR out-of-order - // before start, don't restart - if resTS.IsRestart { - r.logger.Infow( - "rolling back timestamp restart", - "tsBefore", resTS.PreExtendedStart, - "tsAfter", r.timestamp.GetExtendedStart(), - "snBefore", resSN.PreExtendedStart, - "snAfter", r.sequenceNumber.GetExtendedStart(), - ) - r.timestamp.RollbackRestart(resTS.PreExtendedStart) - } - if resSN.IsRestart { - r.logger.Infow( - "rolling back sequence number restart", - "snBefore", resSN.PreExtendedStart, - "snAfter", r.sequenceNumber.GetExtendedStart(), - "tsBefore", resTS.PreExtendedStart, - "tsAfter", r.timestamp.GetExtendedStart(), - ) - r.sequenceNumber.RollbackRestart(resSN.PreExtendedStart) - flowState.IsNotHandled = true - return - } - if -gapSN >= cNumSequenceNumbers/2 { r.logger.Warnw( "large sequence number gap negative", nil, @@ -179,36 +159,6 @@ func (r *RTPStatsReceiver) Update( r.packetsOutOfOrder++ } - if resSN.IsRestart { - r.packetsLost += resSN.PreExtendedStart - resSN.ExtendedVal - - extStartSN := r.sequenceNumber.GetExtendedStart() - for i := uint32(0); i < r.nextSnapshotID-cFirstSnapshotID; i++ { - s := &r.snapshots[i] - if s.extStartSN == resSN.PreExtendedStart { - s.extStartSN = extStartSN - } - } - - r.logger.Infow( - "adjusting start sequence number", - "snBefore", resSN.PreExtendedStart, - "snAfter", resSN.ExtendedVal, - "tsBefore", resTS.PreExtendedStart, - "tsAfter", resTS.ExtendedVal, - ) - } - - if resTS.IsRestart { - r.logger.Infow( - "adjusting start timestamp", - "tsBefore", resTS.PreExtendedStart, - "tsAfter", resTS.ExtendedVal, - "snBefore", resSN.PreExtendedStart, - "snAfter", resSN.ExtendedVal, - ) - } - if r.isInRange(resSN.ExtendedVal, resSN.PreExtendedHighest) { if r.history.IsSet(resSN.ExtendedVal) { r.bytesDuplicate += pktSize diff --git a/pkg/sfu/utils/wraparound.go b/pkg/sfu/utils/wraparound.go index b04b331c2..e37d5ca27 100644 --- a/pkg/sfu/utils/wraparound.go +++ b/pkg/sfu/utils/wraparound.go @@ -26,7 +26,12 @@ type extendedNumber interface { uint32 | uint64 } +type WrapAroundParams struct { + IsRestartAllowed bool +} + type WrapAround[T number, ET extendedNumber] struct { + params WrapAroundParams fullRange ET initialized bool @@ -36,9 +41,10 @@ type WrapAround[T number, ET extendedNumber] struct { extendedHighest ET } -func NewWrapAround[T number, ET extendedNumber]() *WrapAround[T, ET] { +func NewWrapAround[T number, ET extendedNumber](params WrapAroundParams) *WrapAround[T, ET] { var t T return &WrapAround[T, ET]{ + params: params, fullRange: 1 << (unsafe.Sizeof(t) * 8), } } @@ -52,6 +58,7 @@ func (w *WrapAround[T, ET]) Seed(from *WrapAround[T, ET]) { } type WrapAroundUpdateResult[ET extendedNumber] struct { + IsUnhandled bool // when set, other fields are invalid IsRestart bool PreExtendedStart ET // valid only if IsRestart = true PreExtendedHighest ET @@ -140,20 +147,24 @@ func (w *WrapAround[T, ET]) maybeAdjustStart(val T) (result WrapAroundUpdateResu } if val-w.start > T(w.fullRange>>1) { - // out-of-order with existing start => a new start - result.IsRestart = true - if val > w.start { - result.PreExtendedStart = w.fullRange + ET(w.start) - } else { - result.PreExtendedStart = ET(w.start) - } + if w.params.IsRestartAllowed { + // out-of-order with existing start => a new start + result.IsRestart = true + if val > w.start { + result.PreExtendedStart = w.fullRange + ET(w.start) + } else { + result.PreExtendedStart = ET(w.start) + } - if w.isWrapBack(val, w.highest) { - w.cycles = w.fullRange - w.updateExtendedHighest() - cycles = 0 + if w.isWrapBack(val, w.highest) { + w.cycles = w.fullRange + w.updateExtendedHighest() + cycles = 0 + } + w.start = val + } else { + result.IsUnhandled = true } - w.start = val } else { if w.isWrapBack(val, w.highest) { cycles -= w.fullRange diff --git a/pkg/sfu/utils/wraparound_test.go b/pkg/sfu/utils/wraparound_test.go index 55d700650..018e227fc 100644 --- a/pkg/sfu/utils/wraparound_test.go +++ b/pkg/sfu/utils/wraparound_test.go @@ -21,7 +21,7 @@ import ( ) func TestWrapAroundUint16(t *testing.T) { - w := NewWrapAround[uint16, uint32]() + w := NewWrapAround[uint16, uint32](WrapAroundParams{IsRestartAllowed: true}) testCases := []struct { name string input uint16 @@ -194,8 +194,143 @@ func TestWrapAroundUint16(t *testing.T) { } } +func TestWrapAroundUint16NoRestart(t *testing.T) { + w := NewWrapAround[uint16, uint32](WrapAroundParams{IsRestartAllowed: false}) + testCases := []struct { + name string + input uint16 + updated WrapAroundUpdateResult[uint32] + start uint16 + extendedStart uint32 + highest uint16 + extendedHighest uint32 + }{ + // initialize + { + name: "initialize", + input: 10, + updated: WrapAroundUpdateResult[uint32]{ + IsRestart: false, + PreExtendedStart: 0, + PreExtendedHighest: 9, + ExtendedVal: 10, + }, + start: 10, + extendedStart: 10, + highest: 10, + extendedHighest: 10, + }, + // an older number without wrap around should not reset start point + { + name: "no reset start no wrap around", + input: 8, + updated: WrapAroundUpdateResult[uint32]{ + IsUnhandled: true, + // the following fields are not valid when `IsUnhandled = true`, but code fills it in + // and they are filled in here for testing purposes + PreExtendedHighest: 10, + ExtendedVal: 8, + }, + start: 10, + extendedStart: 10, + highest: 10, + extendedHighest: 10, + }, + // an older number with wrap around should not reset start point + { + name: "no reset start wrap around", + input: (1 << 16) - 6, + updated: WrapAroundUpdateResult[uint32]{ + IsUnhandled: true, + PreExtendedHighest: 10, + ExtendedVal: (1 << 16) - 6, + }, + start: 10, + extendedStart: 10, + highest: 10, + extendedHighest: 10, + }, + // yet another older number with wrap around should not reset start point + { + name: "no reset start again", + input: (1 << 16) - 12, + updated: WrapAroundUpdateResult[uint32]{ + IsUnhandled: true, + PreExtendedHighest: 10, + ExtendedVal: (1 << 16) - 12, + }, + start: 10, + extendedStart: 10, + highest: 10, + extendedHighest: 10, + }, + // duplicate should return same as highest + { + name: "duplicate", + input: 10, + updated: WrapAroundUpdateResult[uint32]{ + PreExtendedHighest: 10, + ExtendedVal: 10, + }, + start: 10, + extendedStart: 10, + highest: 10, + extendedHighest: 10, + }, + // a significant jump in order should move highest to that + { + name: "big in-order jump", + input: (1 << 15) - 10, + updated: WrapAroundUpdateResult[uint32]{ + PreExtendedHighest: 10, + ExtendedVal: (1 << 15) - 10, + }, + start: 10, + extendedStart: 10, + highest: (1 << 15) - 10, + extendedHighest: (1 << 15) - 10, + }, + // in-order, should update highest + { + name: "in-order", + input: (1 << 15) + 13, + updated: WrapAroundUpdateResult[uint32]{ + PreExtendedHighest: (1 << 15) - 10, + ExtendedVal: (1 << 15) + 13, + }, + start: 10, + extendedStart: 10, + highest: (1 << 15) + 13, + extendedHighest: (1 << 15) + 13, + }, + // now out-of-order should not reset start as half the range has been seen + { + name: "out-of-order after half range", + input: (1 << 15) - 11, + updated: WrapAroundUpdateResult[uint32]{ + PreExtendedHighest: (1 << 15) + 13, + ExtendedVal: (1 << 15) - 11, + }, + start: 10, + extendedStart: 10, + highest: (1 << 15) + 13, + extendedHighest: (1 << 15) + 13, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.updated, w.Update(tc.input)) + require.Equal(t, tc.start, w.GetStart()) + require.Equal(t, tc.extendedStart, w.GetExtendedStart()) + require.Equal(t, tc.highest, w.GetHighest()) + require.Equal(t, tc.extendedHighest, w.GetExtendedHighest()) + }) + } +} + func TestWrapAroundUint16RollbackRestartAndResetHighest(t *testing.T) { - w := NewWrapAround[uint16, uint64]() + w := NewWrapAround[uint16, uint64](WrapAroundParams{IsRestartAllowed: true}) // initialize w.Update(23) @@ -268,7 +403,7 @@ func TestWrapAroundUint16RollbackRestartAndResetHighest(t *testing.T) { } func TestWrapAroundUint16WrapAroundRestartDuplicate(t *testing.T) { - w := NewWrapAround[uint16, uint64]() + w := NewWrapAround[uint16, uint64](WrapAroundParams{IsRestartAllowed: true}) // initialize w.Update(65534) @@ -314,7 +449,7 @@ func TestWrapAroundUint16WrapAroundRestartDuplicate(t *testing.T) { } func TestWrapAroundUint32(t *testing.T) { - w := NewWrapAround[uint32, uint64]() + w := NewWrapAround[uint32, uint64](WrapAroundParams{IsRestartAllowed: true}) testCases := []struct { name string input uint32