diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 8eb026205..8a091c9d8 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -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 diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index 4f8558512..c3c2e4302 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -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) } } diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index 776434004..94a74f791 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) + 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) diff --git a/pkg/sfu/streamallocator/nacktracker.go b/pkg/sfu/streamallocator/nacktracker.go index 13a53b887..cc91bfe98 100644 --- a/pkg/sfu/streamallocator/nacktracker.go +++ b/pkg/sfu/streamallocator/nacktracker.go @@ -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()) } // ------------------------------------------------ diff --git a/pkg/sfu/streamallocator/trenddetector.go b/pkg/sfu/streamallocator/trenddetector.go index 959b7d9de..232e8878d 100644 --- a/pkg/sfu/streamallocator/trenddetector.go +++ b/pkg/sfu/streamallocator/trenddetector.go @@ -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() {