dandri/livekit
1
0 Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-23 13:02:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Clean up restart a bit. (#2247)

Browse Source
This commit is contained in:
Raja Subramanian authored and GitHub committed 2023-11-17 00:40:00 +05:30
1 parent 7bcd6d4df0
commit c62382c76e
4 files changed
+171 -75
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/sfu/buffer/dependencydescriptorparser.go
+2 -2
View File
@@ -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
}
}
pkg/sfu/buffer/rtpstats_receiver.go
+6 -56
View File
@@ -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
pkg/sfu/utils/wraparound.go
+24 -13
View File
@@ -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
pkg/sfu/utils/wraparound_test.go
+139 -4
View File
@@ -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