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

Ensure sequence number continuity (#1539)

Browse Source 
* Ensure sequence number continuity

When using Go SDK (livekit-cli or egress) as a client,
SFU sends blank frames when audio track is muted to ensure that
Pion OnTrack fires on GoSDK side. That resulted in a huge sequence
number/time stamp jump when the real stream started.

Ensure continuity by creating random sequence number/time stamp when
starting with a blank frames. And when sequence number/time stamp is
initialized using SetLastSnTs, continue sequence if it was already
initialized.

* remove debug
This commit is contained in:
Raja Subramanian
2023-03-22 23:08:26 +05:30
committed by GitHub
parent 23c03f6add
commit 0ea88e4025
2 changed files with 78 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/rtpmunger.go
+27 -9
View File
@@ -2,6 +2,7 @@ package sfu
import (
"fmt"
"math/rand"
"github.com/livekit/protocol/logger"
@@ -41,17 +42,19 @@ type SnTs struct {
// ----------------------------------------------------------------------
type RTPMungerState struct {
LastSN uint16
LastTS uint32
Started bool
LastSN uint16
LastTS uint32
}
func (r RTPMungerState) String() string {
return fmt.Sprintf("RTPMungerState{lastSN: %d, lastTS: %d)", r.LastSN, r.LastTS)
return fmt.Sprintf("RTPMungerState{started: %v, lastSN: %d, lastTS: %d)", r.Started, r.LastSN, r.LastTS)
}
// ----------------------------------------------------------------------
type RTPMungerParams struct {
started bool
highestIncomingSN uint16
lastSN uint16
snOffset uint16
@@ -81,6 +84,7 @@ func NewRTPMunger(logger logger.Logger) *RTPMunger {
func (r *RTPMunger) GetParams() RTPMungerParams {
return RTPMungerParams{
started: r.started,
highestIncomingSN: r.highestIncomingSN,
lastSN: r.lastSN,
snOffset: r.snOffset,
@@ -92,20 +96,28 @@ func (r *RTPMunger) GetParams() RTPMungerParams {
func (r *RTPMunger) GetLast() RTPMungerState {
return RTPMungerState{
LastSN: r.lastSN,
LastTS: r.lastTS,
Started: r.started,
LastSN: r.lastSN,
LastTS: r.lastTS,
}
}
func (r *RTPMunger) SeedLast(state RTPMungerState) {
r.started = state.Started
r.lastSN = state.LastSN
r.lastTS = state.LastTS
}
func (r *RTPMunger) SetLastSnTs(extPkt *buffer.ExtPacket) {
r.highestIncomingSN = extPkt.Packet.SequenceNumber - 1
r.lastSN = extPkt.Packet.SequenceNumber
r.lastTS = extPkt.Packet.Timestamp
if !r.started {
r.lastSN = extPkt.Packet.SequenceNumber
r.lastTS = extPkt.Packet.Timestamp
} else {
r.snOffset = extPkt.Packet.SequenceNumber - r.lastSN - 1
r.tsOffset = extPkt.Packet.Timestamp - r.lastTS - 1
}
r.started = true
}
func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint16, tsAdjust uint32) {
@@ -122,7 +134,7 @@ func (r *RTPMunger) PacketDropped(extPkt *buffer.ExtPacket) {
if r.highestIncomingSN != extPkt.Packet.SequenceNumber {
return
}
r.snOffset += 1
r.snOffset++
r.lastSN = extPkt.Packet.SequenceNumber - r.snOffset
r.snOffsetsWritePtr = (r.snOffsetsWritePtr - 1) & SnOffsetCacheMask
@@ -177,7 +189,7 @@ func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket) (*TranslationPara
// sequence number offset.
if len(extPkt.Packet.Payload) == 0 {
r.highestIncomingSN = extPkt.Packet.SequenceNumber
r.snOffset += 1
r.snOffset++
return &TranslationParamsRTP{
snOrdering: SequenceNumberOrderingContiguous,
@@ -241,6 +253,12 @@ func (r *RTPMunger) UpdateAndGetPaddingSnTs(num int, clockRate uint32, frameRate
tsOffset = 1
}
if !r.started {
r.lastSN = uint16(rand.Intn(1<<14)) + uint16(1<<15) // a random number in third quartile of sequence number space
r.lastTS = uint32(rand.Intn(1<<30)) + uint32(1<<31) // a random number in third quartile of time stamp space
r.started = true
}
vals := make([]SnTs, num)
for i := 0; i < num; i++ {
vals[i].sequenceNumber = r.lastSN + uint16(i) + 1
pkg/sfu/rtpmunger_test.go
+51 -28
View File
@@ -27,14 +27,18 @@ func TestSetLastSnTs(t *testing.T) {
require.NotNil(t, extPkt)
r.SetLastSnTs(extPkt)
require.True(t, r.highestIncomingSN == 23332)
require.True(t, r.lastSN == 23333)
require.True(t, r.lastTS == 0xabcdef)
require.Equal(t, uint16(23332), r.highestIncomingSN)
require.Equal(t, uint16(23333), r.lastSN)
require.Equal(t, uint32(0xabcdef), r.lastTS)
require.Equal(t, uint16(0), r.snOffset)
require.Equal(t, uint32(0), r.tsOffset)
require.True(t, r.started)
// force re-start
r.started = false
params = &testutils.TestExtPacketParams{
SequenceNumber: 0,
SequenceNumber: 43,
Timestamp: 0xabcdef,
SSRC: 0x12345678,
}
@@ -43,11 +47,30 @@ func TestSetLastSnTs(t *testing.T) {
require.NotNil(t, extPkt)
r.SetLastSnTs(extPkt)
require.True(t, r.highestIncomingSN == 65535)
require.True(t, r.lastSN == 0)
require.True(t, r.lastTS == 0xabcdef)
require.Equal(t, uint16(42), r.highestIncomingSN)
require.Equal(t, uint16(43), r.lastSN)
require.Equal(t, uint32(0xabcdef), r.lastTS)
require.Equal(t, uint16(0), r.snOffset)
require.Equal(t, uint32(0), r.tsOffset)
require.True(t, r.started)
// set on a started munger
params = &testutils.TestExtPacketParams{
SequenceNumber: 23457,
Timestamp: 0xabcdef,
SSRC: 0x12345678,
}
extPkt, err = testutils.GetTestExtPacket(params)
require.NoError(t, err)
require.NotNil(t, extPkt)
r.SetLastSnTs(extPkt)
require.Equal(t, uint16(23456), r.highestIncomingSN)
require.Equal(t, uint16(43), r.lastSN)
require.Equal(t, uint32(0xabcdef), r.lastTS)
require.Equal(t, uint16(23413), r.snOffset)
require.Equal(t, uint32(0xffffffff), r.tsOffset)
require.True(t, r.started)
}
func TestUpdateSnTsOffsets(t *testing.T) {
@@ -68,9 +91,9 @@ func TestUpdateSnTsOffsets(t *testing.T) {
}
extPkt, _ = testutils.GetTestExtPacket(params)
r.UpdateSnTsOffsets(extPkt, 1, 1)
require.True(t, r.highestIncomingSN == 33332)
require.True(t, r.lastSN == 23333)
require.True(t, r.lastTS == 0xabcdef)
require.Equal(t, uint16(33332), r.highestIncomingSN)
require.Equal(t, uint16(23333), r.lastSN)
require.Equal(t, uint32(0xabcdef), r.lastTS)
require.Equal(t, uint16(9999), r.snOffset)
require.Equal(t, uint32(0xffffffff), r.tsOffset)
}
@@ -207,8 +230,8 @@ func TestPaddingOnlyPacket(t *testing.T) {
require.Error(t, err)
require.ErrorIs(t, err, ErrPaddingOnlyPacket)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 23333)
require.True(t, r.lastSN == 23333)
require.Equal(t, uint16(23333), r.highestIncomingSN)
require.Equal(t, uint16(23333), r.lastSN)
require.Equal(t, uint16(1), r.snOffset)
// padding only packet with a gap should not report an error
@@ -228,8 +251,8 @@ func TestPaddingOnlyPacket(t *testing.T) {
tp, err = r.UpdateAndGetSnTs(extPkt)
require.NoError(t, err)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 23335)
require.True(t, r.lastSN == 23334)
require.Equal(t, uint16(23335), r.highestIncomingSN)
require.Equal(t, uint16(23334), r.lastSN)
require.Equal(t, uint16(1), r.snOffset)
}
@@ -266,8 +289,8 @@ func TestGapInSequenceNumber(t *testing.T) {
tp, err := r.UpdateAndGetSnTs(extPkt)
require.NoError(t, err)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 1)
require.True(t, r.lastSN == 1)
require.Equal(t, uint16(1), r.highestIncomingSN)
require.Equal(t, uint16(1), r.lastSN)
require.Equal(t, uint16(0), r.snOffset)
// ensure missing sequence numbers got recorded in cache
@@ -294,8 +317,8 @@ func TestGapInSequenceNumber(t *testing.T) {
tp, err = r.UpdateAndGetSnTs(extPkt)
require.ErrorIs(t, err, ErrPaddingOnlyPacket)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 2)
require.True(t, r.lastSN == 1)
require.Equal(t, uint16(2), r.highestIncomingSN)
require.Equal(t, uint16(1), r.lastSN)
require.Equal(t, uint16(1), r.snOffset)
// a packet with a gap should be adding to missing cache
@@ -316,8 +339,8 @@ func TestGapInSequenceNumber(t *testing.T) {
tp, err = r.UpdateAndGetSnTs(extPkt)
require.NoError(t, err)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 4)
require.True(t, r.lastSN == 3)
require.Equal(t, uint16(4), r.highestIncomingSN)
require.Equal(t, uint16(3), r.lastSN)
require.Equal(t, uint16(1), r.snOffset)
// another contiguous padding only packet should be dropped
@@ -335,8 +358,8 @@ func TestGapInSequenceNumber(t *testing.T) {
tp, err = r.UpdateAndGetSnTs(extPkt)
require.ErrorIs(t, err, ErrPaddingOnlyPacket)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 5)
require.True(t, r.lastSN == 3)
require.Equal(t, uint16(5), r.highestIncomingSN)
require.Equal(t, uint16(3), r.lastSN)
require.Equal(t, uint16(2), r.snOffset)
// a packet with a gap should be adding to missing cache
@@ -357,8 +380,8 @@ func TestGapInSequenceNumber(t *testing.T) {
tp, err = r.UpdateAndGetSnTs(extPkt)
require.NoError(t, err)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 7)
require.True(t, r.lastSN == 5)
require.Equal(t, uint16(7), r.highestIncomingSN)
require.Equal(t, uint16(5), r.lastSN)
require.Equal(t, uint16(2), r.snOffset)
// check the missing packets
@@ -378,8 +401,8 @@ func TestGapInSequenceNumber(t *testing.T) {
tp, err = r.UpdateAndGetSnTs(extPkt)
require.NoError(t, err)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 7)
require.True(t, r.lastSN == 5)
require.Equal(t, uint16(7), r.highestIncomingSN)
require.Equal(t, uint16(5), r.lastSN)
require.Equal(t, uint16(2), r.snOffset)
params = &testutils.TestExtPacketParams{
@@ -398,8 +421,8 @@ func TestGapInSequenceNumber(t *testing.T) {
tp, err = r.UpdateAndGetSnTs(extPkt)
require.NoError(t, err)
require.Equal(t, tpExpected, *tp)
require.True(t, r.highestIncomingSN == 7)
require.True(t, r.lastSN == 5)
require.Equal(t, uint16(7), r.highestIncomingSN)
require.Equal(t, uint16(5), r.lastSN)
require.Equal(t, uint16(2), r.snOffset)
}