From 0ea88e40250b436b42b670974e8b6a60659cda91 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 22 Mar 2023 23:08:26 +0530 Subject: [PATCH] Ensure sequence number continuity (#1539) * 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 --- pkg/sfu/rtpmunger.go | 36 +++++++++++++----- pkg/sfu/rtpmunger_test.go | 79 +++++++++++++++++++++++++-------------- 2 files changed, 78 insertions(+), 37 deletions(-) diff --git a/pkg/sfu/rtpmunger.go b/pkg/sfu/rtpmunger.go index d42a4fe5f..1cd27d745 100644 --- a/pkg/sfu/rtpmunger.go +++ b/pkg/sfu/rtpmunger.go @@ -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 diff --git a/pkg/sfu/rtpmunger_test.go b/pkg/sfu/rtpmunger_test.go index faa68efe9..1d59759ab 100644 --- a/pkg/sfu/rtpmunger_test.go +++ b/pkg/sfu/rtpmunger_test.go @@ -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) }