diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 3c7aeeb10..ff5c4e675 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -31,7 +31,6 @@ type pendingPacket struct { } type ExtPacket struct { - Head bool Arrival int64 Packet *rtp.Packet Payload interface{} @@ -364,10 +363,10 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) { return } - flowState := b.updateStreamState(&p, arrivalTime) + b.updateStreamState(&p, arrivalTime) b.processHeaderExtensions(&p, arrivalTime) - ep := b.getExtPacket(pb, &p, arrivalTime, flowState.IsHighestSN) + ep := b.getExtPacket(pb, &p, arrivalTime) if ep == nil { return } @@ -378,7 +377,7 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) { b.doReports(arrivalTime) } -func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime int64) RTPFlowState { +func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime int64) { flowState := b.rtpStats.Update(&p.Header, len(p.Payload), int(p.PaddingSize), arrivalTime) if b.nacker != nil { @@ -390,8 +389,6 @@ func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime int64) RTPFlowStat } } } - - return flowState } func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) { @@ -424,9 +421,8 @@ func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) { } } -func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTime int64, isHighestSN bool) *ExtPacket { +func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTime int64) *ExtPacket { ep := &ExtPacket{ - Head: isHighestSN, Packet: rtpPacket, Arrival: arrivalTime, RawPacket: rawPacket, diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 6223dc714..fab09a60f 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -24,7 +24,6 @@ const ( ) type RTPFlowState struct { - IsHighestSN bool HasLoss bool LossStartInclusive uint16 LossEndExclusive uint16 @@ -244,7 +243,6 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa // in-order default: - flowState.IsHighestSN = true if diff > 1 { flowState.HasLoss = true flowState.LossStartInclusive = r.highestSN + 1 diff --git a/pkg/sfu/buffer/rtpstats_test.go b/pkg/sfu/buffer/rtpstats_test.go index 584818cbe..1fbbc7a4e 100644 --- a/pkg/sfu/buffer/rtpstats_test.go +++ b/pkg/sfu/buffer/rtpstats_test.go @@ -71,7 +71,6 @@ func TestRTPStats_Update(t *testing.T) { timestamp := uint32(rand.Float64() * float64(1<<32)) packet := getPacket(sequenceNumber, timestamp, 1000) flowState := r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) - require.True(t, flowState.IsHighestSN) require.False(t, flowState.HasLoss) require.True(t, r.initialized) require.Equal(t, sequenceNumber, r.highestSN) @@ -82,7 +81,6 @@ func TestRTPStats_Update(t *testing.T) { timestamp += 3000 packet = getPacket(sequenceNumber, timestamp, 1000) flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) - require.True(t, flowState.IsHighestSN) require.False(t, flowState.HasLoss) require.Equal(t, sequenceNumber, r.highestSN) require.Equal(t, timestamp, r.highestTS) @@ -90,7 +88,6 @@ func TestRTPStats_Update(t *testing.T) { // out-of-order packet = getPacket(sequenceNumber-10, timestamp-30000, 1000) flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) - require.False(t, flowState.IsHighestSN) require.False(t, flowState.HasLoss) require.Equal(t, sequenceNumber, r.highestSN) require.Equal(t, timestamp, r.highestTS) @@ -100,7 +97,6 @@ func TestRTPStats_Update(t *testing.T) { // duplicate packet = getPacket(sequenceNumber-10, timestamp-30000, 1000) flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) - require.False(t, flowState.IsHighestSN) require.False(t, flowState.HasLoss) require.Equal(t, sequenceNumber, r.highestSN) require.Equal(t, timestamp, r.highestTS) @@ -112,7 +108,6 @@ func TestRTPStats_Update(t *testing.T) { timestamp += 30000 packet = getPacket(sequenceNumber, timestamp, 1000) flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) - require.True(t, flowState.IsHighestSN) require.True(t, flowState.HasLoss) require.Equal(t, sequenceNumber-9, flowState.LossStartInclusive) require.Equal(t, sequenceNumber, flowState.LossEndExclusive) @@ -121,7 +116,6 @@ func TestRTPStats_Update(t *testing.T) { // out-of-order should decrement number of lost packets packet = getPacket(sequenceNumber-15, timestamp-45000, 1000) flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano()) - require.False(t, flowState.IsHighestSN) require.False(t, flowState.HasLoss) require.Equal(t, sequenceNumber, r.highestSN) require.Equal(t, timestamp, r.highestTS) diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 3c96f7a9c..0e37150f8 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -791,7 +791,6 @@ func TestForwarderGetTranslationParamsAudio(t *testing.T) { f := newForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio) params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -840,7 +839,6 @@ func TestForwarderGetTranslationParamsAudio(t *testing.T) { // padding only packet in order should be dropped params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23334, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -856,7 +854,6 @@ func TestForwarderGetTranslationParamsAudio(t *testing.T) { // in order packet should be forwarded params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23335, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -877,7 +874,6 @@ func TestForwarderGetTranslationParamsAudio(t *testing.T) { // padding only packet after a gap should be forwarded params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23337, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -897,7 +893,6 @@ func TestForwarderGetTranslationParamsAudio(t *testing.T) { // out-of-order should be forwarded using cache params = &testutils.TestExtPacketParams{ - IsHead: false, SequenceNumber: 23336, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -918,7 +913,6 @@ func TestForwarderGetTranslationParamsAudio(t *testing.T) { // switching source should lock onto the new source, but sequence number should be contiguous params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 123, Timestamp: 0xfedcba, SSRC: 0x87654321, @@ -943,7 +937,6 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1059,7 +1052,6 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { // padding only packet in order should be dropped params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23334, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1074,7 +1066,6 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { // in order packet should be forwarded params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23335, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1111,7 +1102,6 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { // temporal layer higher than target, should be dropped params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23336, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1142,7 +1132,6 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { // RTP sequence number and VP8 picture id should be contiguous after dropping higher temporal layer picture params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23337, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1194,7 +1183,6 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { // padding only packet after a gap should be forwarded params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23339, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1214,7 +1202,6 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { // out-of-order should be forwarded using cache, even if it is padding only params = &testutils.TestExtPacketParams{ - IsHead: false, SequenceNumber: 23338, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1240,7 +1227,6 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { } params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 123, Timestamp: 0xfedcba, SSRC: 0x87654321, @@ -1297,7 +1283,6 @@ func TestForwardGetSnTsForPadding(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1365,7 +1350,6 @@ func TestForwardGetSnTsForBlankFrames(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -1436,7 +1420,6 @@ func TestForwardGetPaddingVP8(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, diff --git a/pkg/sfu/rtpmunger.go b/pkg/sfu/rtpmunger.go index 8d9ae098b..a24c66745 100644 --- a/pkg/sfu/rtpmunger.go +++ b/pkg/sfu/rtpmunger.go @@ -92,11 +92,9 @@ func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint16, } func (r *RTPMunger) PacketDropped(extPkt *buffer.ExtPacket) { - if !extPkt.Head { + if r.highestIncomingSN != extPkt.Packet.SequenceNumber { return } - - r.highestIncomingSN = extPkt.Packet.SequenceNumber r.snOffset += 1 r.lastSN = extPkt.Packet.SequenceNumber - r.snOffset @@ -109,7 +107,8 @@ func (r *RTPMunger) PacketDropped(extPkt *buffer.ExtPacket) { func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket) (*TranslationParamsRTP, error) { // if out-of-order, look up sequence number offset cache - if !extPkt.Head { + diff := extPkt.Packet.SequenceNumber - r.highestIncomingSN + if diff > (1 << 15) { snOffset, isValid := r.getSnOffset(extPkt.Packet.SequenceNumber) if !isValid { return &TranslationParamsRTP{ @@ -126,7 +125,6 @@ func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket) (*TranslationPara ordering := SequenceNumberOrderingContiguous - diff := extPkt.Packet.SequenceNumber - r.highestIncomingSN if diff > 1 { ordering = SequenceNumberOrderingGap } else { diff --git a/pkg/sfu/rtpmunger_test.go b/pkg/sfu/rtpmunger_test.go index d50d13668..cff7fa06c 100644 --- a/pkg/sfu/rtpmunger_test.go +++ b/pkg/sfu/rtpmunger_test.go @@ -105,14 +105,13 @@ func TestPacketDropped(t *testing.T) { // drop a head packet and check offset increases params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 44444, Timestamp: 0xabcdef, SSRC: 0x12345678, } extPkt, _ = testutils.GetTestExtPacket(params) + r.highestIncomingSN = 44444 r.PacketDropped(extPkt) - require.Equal(t, r.highestIncomingSN, uint16(44444)) require.Equal(t, r.lastSN, uint16(44443)) require.Equal(t, uint16(1), r.snOffset) } @@ -124,15 +123,18 @@ func TestOutOfOrderSequenceNumber(t *testing.T) { SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, + PayloadSize: 10, } extPkt, _ := testutils.GetTestExtPacket(params) r.SetLastSnTs(extPkt) + r.UpdateAndGetSnTs(extPkt) // out-of-order sequence number not in the missing sequence number cache params = &testutils.TestExtPacketParams{ SequenceNumber: 23332, Timestamp: 0xabcdef, SSRC: 0x12345678, + PayloadSize: 10, } extPkt, _ = testutils.GetTestExtPacket(params) @@ -164,7 +166,6 @@ func TestDuplicateSequenceNumber(t *testing.T) { r := newRTPMunger() params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -190,7 +191,6 @@ func TestPaddingOnlyPacket(t *testing.T) { r := newRTPMunger() params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -213,7 +213,6 @@ func TestPaddingOnlyPacket(t *testing.T) { // padding only packet with a gap should not report an error params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23335, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -238,7 +237,6 @@ func TestGapInSequenceNumber(t *testing.T) { r := newRTPMunger() params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 65533, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -252,7 +250,6 @@ func TestGapInSequenceNumber(t *testing.T) { // three lost packets params = &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 1, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -287,7 +284,6 @@ func TestUpdateAndGetPaddingSnTs(t *testing.T) { r := newRTPMunger() params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -333,7 +329,6 @@ func TestIsOnFrameBoundary(t *testing.T) { r := newRTPMunger() params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -349,7 +344,6 @@ func TestIsOnFrameBoundary(t *testing.T) { // packet with RTP marker params = &testutils.TestExtPacketParams{ - IsHead: true, SetMarker: true, SequenceNumber: 23334, Timestamp: 0xabcdef, diff --git a/pkg/sfu/testutils/data.go b/pkg/sfu/testutils/data.go index 8c65f6dd9..a3131eeee 100644 --- a/pkg/sfu/testutils/data.go +++ b/pkg/sfu/testutils/data.go @@ -11,7 +11,6 @@ import ( type TestExtPacketParams struct { SetMarker bool - IsHead bool IsKeyFrame bool PayloadType uint8 SequenceNumber uint16 @@ -45,7 +44,6 @@ func GetTestExtPacket(params *TestExtPacketParams) (*buffer.ExtPacket, error) { } ep := &buffer.ExtPacket{ - Head: params.IsHead, Arrival: params.ArrivalTime, Packet: &packet, KeyFrame: params.IsKeyFrame, diff --git a/pkg/sfu/vp8munger_test.go b/pkg/sfu/vp8munger_test.go index 1e93f52c8..e919f0e3d 100644 --- a/pkg/sfu/vp8munger_test.go +++ b/pkg/sfu/vp8munger_test.go @@ -321,7 +321,6 @@ func TestGapInSequenceNumberSamePicture(t *testing.T) { v := newVP8Munger() params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 65533, Timestamp: 0xabcdef, SSRC: 0x12345678, @@ -397,7 +396,6 @@ func TestUpdateAndGetPadding(t *testing.T) { v := newVP8Munger() params := &testutils.TestExtPacketParams{ - IsHead: true, SequenceNumber: 23333, Timestamp: 0xabcdef, SSRC: 0x12345678,