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

Safe access of sequencer (#1625)

Browse Source 
* log some NACKs

* split out NACK tracker

* remove debug

* debug

* Sequencer safety

* Clean up
This commit is contained in:
Raja Subramanian authored and GitHub committed 2023-04-17 13:02:24 +05:30
1 parent d9cd07c4b1
commit 5b34d754e0
5 files changed
+57 -32
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/sfu/downtrack.go
+13 -13
View File
@@ -563,12 +563,15 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
payload = d.translateVP8PacketTo(extPkt.Packet, &incomingVP8, tp.codecBytes, pool)
}
var meta *packetMeta
if d.sequencer != nil {
meta = d.sequencer.push(extPkt.Packet.SequenceNumber, tp.rtp.sequenceNumber, tp.rtp.timestamp, int8(layer))
if meta != nil {
meta.codecBytes = append(meta.codecBytes, tp.codecBytes...)
}
d.sequencer.push(
extPkt.Packet.SequenceNumber,
tp.rtp.sequenceNumber,
tp.rtp.timestamp,
int8(layer),
tp.codecBytes,
tp.ddBytes,
)
}
hdr, err := d.getTranslatedRTPHeader(extPkt, tp)
@@ -577,10 +580,6 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
return err
}
if meta != nil && d.dependencyDescriptorID != 0 {
meta.ddBytes = append(meta.ddBytes, tp.ddBytes...)
}
_, err = d.writeStream.WriteRTP(hdr, payload)
if err != nil {
if errors.Is(err, io.ErrClosedPipe) {
@@ -603,7 +602,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
if extPkt.KeyFrame {
d.isNACKThrottled.Store(false)
d.rtpStats.UpdateKeyFrame(1)
d.logger.Debugw("forwarding key frame", "layer", layer)
d.logger.Debugw("forwarding key frame", "layer", layer, "rtpsn", hdr.SequenceNumber, "rtpts", hdr.Timestamp)
}
if tp.isSwitchingToRequestSpatial {
@@ -1443,9 +1442,10 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
continue
}
translatedVP8 := meta.codecBytes
pool = PacketFactory.Get().(*[]byte)
payload = d.translateVP8PacketTo(&pkt, &incomingVP8, translatedVP8, pool)
if len(meta.codecBytes) != 0 {
pool = PacketFactory.Get().(*[]byte)
payload = d.translateVP8PacketTo(&pkt, &incomingVP8, meta.codecBytes, pool)
}
}
var extraExtensions []extensionData
pkg/sfu/sequencer.go
+11 -7
View File
@@ -92,13 +92,13 @@ func (s *sequencer) setRTT(rtt uint32) {
}
}
func (s *sequencer) push(sn, offSn uint16, timeStamp uint32, layer int8) *packetMeta {
func (s *sequencer) push(sn, offSn uint16, timeStamp uint32, layer int8, codecBytes []byte, ddBytes []byte) {
s.Lock()
defer s.Unlock()
slot, isValid := s.getSlot(offSn)
if !isValid {
return nil
return
}
s.meta[s.metaWritePtr] = packetMeta{
@@ -106,6 +106,8 @@ func (s *sequencer) push(sn, offSn uint16, timeStamp uint32, layer int8) *packet
targetSeqNo: offSn,
timestamp: timeStamp,
layer: layer,
codecBytes: append([]byte{}, codecBytes...),
ddBytes: append([]byte{}, ddBytes...),
}
s.seq[slot] = &s.meta[s.metaWritePtr]
@@ -114,8 +116,6 @@ func (s *sequencer) push(sn, offSn uint16, timeStamp uint32, layer int8) *packet
if s.metaWritePtr >= len(s.meta) {
s.metaWritePtr -= len(s.meta)
}
return s.seq[slot]
}
func (s *sequencer) pushPadding(offSn uint16) {
@@ -168,11 +168,11 @@ func (s *sequencer) getSlot(offSn uint16) (int, bool) {
return s.wrap(slot), true
}
func (s *sequencer) getPacketsMeta(seqNo []uint16) []*packetMeta {
func (s *sequencer) getPacketsMeta(seqNo []uint16) []packetMeta {
s.Lock()
defer s.Unlock()
meta := make([]*packetMeta, 0, len(seqNo))
meta := make([]packetMeta, 0, len(seqNo))
refTime := uint32(time.Now().UnixNano()/1e6 - s.startTime)
for _, sn := range seqNo {
diff := s.headSN - sn
@@ -190,7 +190,11 @@ func (s *sequencer) getPacketsMeta(seqNo []uint16) []*packetMeta {
if seq.lastNack == 0 || refTime-seq.lastNack > uint32(math.Min(float64(ignoreRetransmission), float64(2*s.rtt))) {
seq.nacked++
seq.lastNack = refTime
meta = append(meta, seq)
pm := *seq
pm.codecBytes = append([]byte{}, seq.codecBytes...)
pm.ddBytes = append([]byte{}, seq.ddBytes...)
meta = append(meta, pm)
}
}
pkg/sfu/sequencer_test.go
+25 -6
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)
seq.push(i, i+off, 123, 2, nil, nil)
}
// send the last two out-of-order
seq.push(519, 519+off, 123, 2)
seq.push(518, 518+off, 123, 2)
seq.push(519, 519+off, 123, 2, nil, nil)
seq.push(518, 518+off, 123, 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)
seq.push(521, 521+off, 123, 1, nil, nil)
m := seq.getPacketsMeta([]uint16{521 + off})
require.Equal(t, 1, len(m))
seq.push(505, 505+off, 123, 1)
seq.push(505, 505+off, 123, 1, nil, nil)
m = seq.getPacketsMeta([]uint16{505 + off})
require.Equal(t, 1, len(m))
}
@@ -58,6 +58,10 @@ func Test_sequencer_getNACKSeqNo(t *testing.T) {
input []uint16
padding []uint16
offset uint16
codecBytesOdd []byte
codecBytesEven []byte
ddBytesOdd []byte
ddBytesEven []byte
}
tests := []struct {
@@ -72,6 +76,10 @@ func Test_sequencer_getNACKSeqNo(t *testing.T) {
input: []uint16{2, 3, 4, 7, 8, 11},
padding: []uint16{9, 10},
offset: 5,
codecBytesOdd: []byte{1, 2, 3, 4},
codecBytesEven: []byte{5, 6, 7},
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},
@@ -85,7 +93,11 @@ func Test_sequencer_getNACKSeqNo(t *testing.T) {
n := newSequencer(5, 10, logger.GetLogger())
for _, i := range tt.fields.input {
n.push(i, i+tt.fields.offset, 123, 3)
if i % 2 == 0 {
n.push(i, i+tt.fields.offset, 123, 3, tt.fields.codecBytesEven, tt.fields.ddBytesEven)
} else {
n.push(i, i+tt.fields.offset, 123, 3, tt.fields.codecBytesOdd, tt.fields.ddBytesOdd)
}
}
for _, i := range tt.fields.padding {
n.pushPadding(i + tt.fields.offset)
@@ -95,6 +107,13 @@ func Test_sequencer_getNACKSeqNo(t *testing.T) {
var got []uint16
for _, sn := range g {
got = append(got, sn.sourceSeqNo)
if sn.sourceSeqNo % 2 == 0 {
require.Equal(t, tt.fields.codecBytesEven, sn.codecBytes)
require.Equal(t, tt.fields.ddBytesEven, sn.ddBytes)
} else {
require.Equal(t, tt.fields.codecBytesOdd, sn.codecBytes)
require.Equal(t, tt.fields.ddBytesOdd, sn.ddBytes)
}
}
if !reflect.DeepEqual(got, tt.want) {
t.Errorf("getPacketsMeta() = %v, want %v", got, tt.want)
pkg/sfu/streamallocator/nacktracker.go
+4 -2
View File
@@ -77,9 +77,11 @@ func (n *NackTracker) IsTriggered() bool {
func (n *NackTracker) ToString() string {
window := ""
if !n.windowStartTime.IsZero() {
window = fmt.Sprintf("t: %+v|%+v", n.windowStartTime, time.Since(n.windowStartTime))
now := time.Now()
elapsed := now.Sub(n.windowStartTime).Seconds()
window = fmt.Sprintf("t: %+v|%+v|%.2fs", n.windowStartTime.Format(time.UnixDate), now.Format(time.UnixDate), elapsed)
}
return fmt.Sprintf("n: %s, t: %s, p: %d, rn: %d, rn/p: %f", n.params.Name, window, n.packets, n.repeatedNacks, n.GetRatio())
return fmt.Sprintf("n: %s, t: %s, p: %d, rn: %d, rn/p: %.2f", n.params.Name, window, n.packets, n.repeatedNacks, n.GetRatio())
}
// ------------------------------------------------
pkg/sfu/streamallocator/trenddetector.go
+4 -4
View File
@@ -109,10 +109,10 @@ func (t *TrendDetector) GetDirection() TrendDirection {
func (t *TrendDetector) ToString() string {
now := time.Now()
elapsed := now.Sub(t.startTime).Seconds()
str := fmt.Sprintf("n: %s", t.params.Name)
str += fmt.Sprintf(", t: %+v|%+v|%.2fs", t.startTime.Format(time.UnixDate), now.Format(time.UnixDate), elapsed)
str += fmt.Sprintf(", v: %d|%d|%d|%+v|%.2f", t.numSamples, t.lowestValue, t.highestValue, t.values, kendallsTau(t.values))
return str
return fmt.Sprintf("n: %s, t: %+v|%+v|%.2fs, v: %d|%d|%d|%+v|%.2f",
t.params.Name,
t.startTime.Format(time.UnixDate), now.Format(time.UnixDate), elapsed,
t.numSamples, t.lowestValue, t.highestValue, t.values, kendallsTau(t.values))
}
func (t *TrendDetector) updateDirection() {