From 887f6580ec5319b4682ee31861320c38b3489cc1 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 27 Jul 2023 17:08:14 +0530 Subject: [PATCH] Cache marker in sequencer and use it while retransmit. (#1912) With SVC codecs, input marker and fowarded marker could be different. So, cache it in sequence and use it on retransmit. @cndderrauber - this could have affected SVC under packet loss. --- pkg/sfu/downtrack.go | 24 +++++++++++--------- pkg/sfu/sequencer.go | 12 +++++++++- pkg/sfu/sequencer_test.go | 48 ++++++++++++++++++++++----------------- 3 files changed, 51 insertions(+), 33 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 4147b1650..0ad86315c 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -638,17 +638,6 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { copy(payload, extPkt.Packet.Payload) } - if d.sequencer != nil { - d.sequencer.push( - extPkt.Packet.SequenceNumber, - tp.rtp.sequenceNumber, - tp.rtp.timestamp, - int8(layer), - tp.codecBytes, - tp.ddBytes, - ) - } - hdr, err := d.getTranslatedRTPHeader(extPkt, tp) if err != nil { d.logger.Errorw("write rtp packet failed", err) @@ -658,6 +647,18 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { return err } + if d.sequencer != nil { + d.sequencer.push( + extPkt.Packet.SequenceNumber, + tp.rtp.sequenceNumber, + tp.rtp.timestamp, + hdr.Marker, + int8(layer), + tp.codecBytes, + tp.ddBytes, + ) + } + d.pacer.Enqueue(pacer.Packet{ Header: hdr, Extensions: []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: tp.ddBytes}}, @@ -1516,6 +1517,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { d.logger.Errorw("unmarshalling rtp packet failed in retransmit", err) continue } + pkt.Header.Marker = meta.marker pkt.Header.SequenceNumber = meta.targetSeqNo pkt.Header.Timestamp = meta.timestamp pkt.Header.SSRC = d.ssrc diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index 4a8b31f9c..333fd93ff 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -39,6 +39,8 @@ type packetMeta struct { // Modified timestamp for current associated // down track. timestamp uint32 + // Modified marker + marker bool // The last time this packet was nack requested. // Sometimes clients request the same packet more than once, so keep // track of the requested packets helps to avoid writing multiple times @@ -93,7 +95,14 @@ func (s *sequencer) setRTT(rtt uint32) { } } -func (s *sequencer) push(sn, offSn uint16, timeStamp uint32, layer int8, codecBytes []byte, ddBytes []byte) { +func (s *sequencer) push( + sn, offSn uint16, + timeStamp uint32, + marker bool, + layer int8, + codecBytes []byte, + ddBytes []byte, +) { s.Lock() defer s.Unlock() @@ -106,6 +115,7 @@ func (s *sequencer) push(sn, offSn uint16, timeStamp uint32, layer int8, codecBy sourceSeqNo: sn, targetSeqNo: offSn, timestamp: timeStamp, + marker: marker, layer: layer, codecBytes: append([]byte{}, codecBytes...), ddBytes: append([]byte{}, ddBytes...), diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index 94a74f791..87434bf03 100644 --- a/pkg/sfu/sequencer_test.go +++ b/pkg/sfu/sequencer_test.go @@ -15,11 +15,11 @@ func Test_sequencer(t *testing.T) { off := uint16(15) for i := uint16(1); i < 518; i++ { - seq.push(i, i+off, 123, 2, nil, nil) + seq.push(i, i+off, 123, true, 2, nil, nil) } // send the last two out-of-order - seq.push(519, 519+off, 123, 2, nil, nil) - seq.push(518, 518+off, 123, 2, nil, nil) + seq.push(519, 519+off, 123, false, 2, nil, nil) + seq.push(518, 518+off, 123, true, 2, nil, nil) time.Sleep(60 * time.Millisecond) req := []uint16{57, 58, 62, 63, 513, 514, 515, 516, 517} @@ -41,11 +41,11 @@ func Test_sequencer(t *testing.T) { require.Equal(t, val.layer, int8(2)) } - seq.push(521, 521+off, 123, 1, nil, nil) + seq.push(521, 521+off, 123, true, 1, nil, nil) m := seq.getPacketsMeta([]uint16{521 + off}) require.Equal(t, 1, len(m)) - seq.push(505, 505+off, 123, 1, nil, nil) + seq.push(505, 505+off, 123, false, 1, nil, nil) m = seq.getPacketsMeta([]uint16{505 + off}) require.Equal(t, 1, len(m)) } @@ -55,13 +55,15 @@ func Test_sequencer_getNACKSeqNo(t *testing.T) { seqNo []uint16 } type fields struct { - input []uint16 - padding []uint16 - offset uint16 - codecBytesOdd []byte + input []uint16 + padding []uint16 + offset uint16 + markerOdd bool + markerEven bool + codecBytesOdd []byte codecBytesEven []byte - ddBytesOdd []byte - ddBytesEven []byte + ddBytesOdd []byte + ddBytesEven []byte } tests := []struct { @@ -73,13 +75,15 @@ func Test_sequencer_getNACKSeqNo(t *testing.T) { { name: "Should get correct seq numbers", fields: fields{ - input: []uint16{2, 3, 4, 7, 8, 11}, - padding: []uint16{9, 10}, - offset: 5, - codecBytesOdd: []byte{1, 2, 3, 4}, + input: []uint16{2, 3, 4, 7, 8, 11}, + padding: []uint16{9, 10}, + offset: 5, + markerOdd: true, + markerEven: false, + codecBytesOdd: []byte{1, 2, 3, 4}, codecBytesEven: []byte{5, 6, 7}, - ddBytesOdd: []byte{8, 9, 10}, - ddBytesEven: []byte{11, 12}, + ddBytesOdd: []byte{8, 9, 10}, + ddBytesEven: []byte{11, 12}, }, args: args{ seqNo: []uint16{4 + 5, 5 + 5, 8 + 5, 9 + 5, 10 + 5, 11 + 5}, @@ -93,10 +97,10 @@ func Test_sequencer_getNACKSeqNo(t *testing.T) { n := newSequencer(5, 10, logger.GetLogger()) for _, i := range tt.fields.input { - if i % 2 == 0 { - n.push(i, i+tt.fields.offset, 123, 3, tt.fields.codecBytesEven, tt.fields.ddBytesEven) + if i%2 == 0 { + n.push(i, i+tt.fields.offset, 123, tt.fields.markerEven, 3, tt.fields.codecBytesEven, tt.fields.ddBytesEven) } else { - n.push(i, i+tt.fields.offset, 123, 3, tt.fields.codecBytesOdd, tt.fields.ddBytesOdd) + n.push(i, i+tt.fields.offset, 123, tt.fields.markerOdd, 3, tt.fields.codecBytesOdd, tt.fields.ddBytesOdd) } } for _, i := range tt.fields.padding { @@ -107,10 +111,12 @@ func Test_sequencer_getNACKSeqNo(t *testing.T) { var got []uint16 for _, sn := range g { got = append(got, sn.sourceSeqNo) - if sn.sourceSeqNo % 2 == 0 { + if sn.sourceSeqNo%2 == 0 { + require.Equal(t, tt.fields.markerEven, sn.marker) require.Equal(t, tt.fields.codecBytesEven, sn.codecBytes) require.Equal(t, tt.fields.ddBytesEven, sn.ddBytes) } else { + require.Equal(t, tt.fields.markerOdd, sn.marker) require.Equal(t, tt.fields.codecBytesOdd, sn.codecBytes) require.Equal(t, tt.fields.ddBytesOdd, sn.ddBytes) }