Remove Head field from ExtPacket structure. (#662)

* Remove `Head` field from `ExtPacket` structure.

Although we do not intend to, but if packets get out-of-order
in the forwarding path (maybe reading in multiple goroutines
or using some worker pool to distribute packets), the `Head`
indicator could lead to wrong behaviour. It is possible that
at the receiver, the order is
- Seq Num N, Head = true
- N + 1, Head = true

If the forwarding path sees `N + 1` first, the Head flag
when it sees `N` packet is incorrect and will lead to incorrect
behaviour.

The alternative check is very simple. So, remove `Head` flag.

* Remove unused field

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,

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

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)

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,

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 {

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,

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,

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,