dandri/livekit
1
0 Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-23 15:12:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cache marker in sequencer and use it while retransmit. (#1912)

Browse Source 
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.
This commit is contained in:
Raja Subramanian authored and GitHub committed 2023-07-27 17:08:14 +05:30
1 parent fc7d4bd01e
commit 887f6580ec
3 files changed
+51 -33
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/sfu/downtrack.go
+13 -11
View File
@@ -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
pkg/sfu/sequencer.go
+11 -1
View File
@@ -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...),
pkg/sfu/sequencer_test.go
+27 -21
View File
@@ -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)
}