From 8b16db227070ab0592140f47dd78da548723f1b4 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 26 Oct 2023 21:07:36 +0530 Subject: [PATCH] Log PLI requests. (#2194) A few things - Log PLI requests from client. - Pass in marker to RTP munger as SVC can insert marker. - Adjusting first packet time should be aware of SVC as there is single stream in SVC --- pkg/sfu/downtrack.go | 12 ++++++----- pkg/sfu/forwarder.go | 4 ++-- pkg/sfu/receiver.go | 2 +- pkg/sfu/rtpmunger.go | 39 ++++++++++++++++++++++----------- pkg/sfu/rtpmunger_test.go | 45 +++++++++++++++++++-------------------- 5 files changed, 59 insertions(+), 43 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 86188797b..a6498dcd4 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -54,7 +54,7 @@ type TrackSender interface { ID() string SubscriberID() livekit.ParticipantID TrackInfoAvailable() - HandleRTCPSenderReportData(payloadType webrtc.PayloadType, layer int32, srData *buffer.RTCPSenderReportData) error + HandleRTCPSenderReportData(payloadType webrtc.PayloadType, isSVC bool, layer int32, srData *buffer.RTCPSenderReportData) error } // ------------------------------------------------------------------- @@ -1463,9 +1463,11 @@ func (d *DownTrack) handleRTCP(bytes []byte) { pliOnce := true sendPliOnce := func() { + _, layer := d.forwarder.CheckSync() + isAnyMuted := d.forwarder.IsAnyMuted() + d.params.Logger.Debugw("received PLI/FIR RTCP", "layer", layer, "isAnyMuted", isAnyMuted) if pliOnce { - _, layer := d.forwarder.CheckSync() - if layer != buffer.InvalidLayerSpatial && !d.forwarder.IsAnyMuted() { + if layer != buffer.InvalidLayerSpatial && !isAnyMuted { d.params.Logger.Debugw("sending PLI RTCP", "layer", layer) d.params.Receiver.SendPLI(layer, false) d.isNACKThrottled.Store(true) @@ -1891,8 +1893,8 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() { } } -func (d *DownTrack) HandleRTCPSenderReportData(_payloadType webrtc.PayloadType, layer int32, srData *buffer.RTCPSenderReportData) error { - if layer == d.forwarder.GetReferenceLayerSpatial() && srData != nil { +func (d *DownTrack) HandleRTCPSenderReportData(_payloadType webrtc.PayloadType, isSVC bool, layer int32, srData *buffer.RTCPSenderReportData) error { + if (layer == d.forwarder.GetReferenceLayerSpatial() || (layer == 0 && isSVC)) && srData != nil { d.rtpStats.MaybeAdjustFirstPacketTime(srData.RTPTimestamp + uint32(d.forwarder.GetReferenceTimestampOffset())) } return nil diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 887c48566..49c955291 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -1653,7 +1653,7 @@ func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer i f.lastSSRC = extPkt.Packet.SSRC } - tpRTP, err := f.rtpMunger.UpdateAndGetSnTs(extPkt) + tpRTP, err := f.rtpMunger.UpdateAndGetSnTs(extPkt, tp.marker) if err != nil { tp.shouldDrop = true if err == ErrPaddingOnlyPacket || err == ErrDuplicatePacket || err == ErrOutOfOrderSequenceNumberCacheMiss { @@ -1691,7 +1691,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in tp.shouldDrop = true if f.started && result.IsRelevant { // call to update highest incoming sequence number and other internal structures - if tpRTP, err := f.rtpMunger.UpdateAndGetSnTs(extPkt); err == nil && tpRTP.snOrdering == SequenceNumberOrderingContiguous { + if tpRTP, err := f.rtpMunger.UpdateAndGetSnTs(extPkt, result.RTPMarker); err == nil && tpRTP.snOrdering == SequenceNumberOrderingContiguous { f.rtpMunger.PacketDropped(extPkt) } } diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 560a9246d..d07a50615 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -342,7 +342,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff w.streamTrackerManager.SetRTCPSenderReportData(layer, srFirst, srNewest) w.downTrackSpreader.Broadcast(func(dt TrackSender) { - _ = dt.HandleRTCPSenderReportData(w.codec.PayloadType, layer, srNewest) + _ = dt.HandleRTCPSenderReportData(w.codec.PayloadType, w.isSVC, layer, srNewest) }) }) diff --git a/pkg/sfu/rtpmunger.go b/pkg/sfu/rtpmunger.go index 1fcda126e..467ec2b3f 100644 --- a/pkg/sfu/rtpmunger.go +++ b/pkg/sfu/rtpmunger.go @@ -51,14 +51,21 @@ type SnTs struct { // ---------------------------------------------------------------------- type RTPMungerState struct { - ExtLastSN uint64 - ExtSecondLastSN uint64 - ExtLastTS uint64 - ExtSecondLastTS uint64 + ExtLastSN uint64 + ExtSecondLastSN uint64 + ExtLastTS uint64 + ExtSecondLastTS uint64 + LastMarker bool + SecondLastMarker bool } func (r RTPMungerState) String() string { - return fmt.Sprintf("RTPMungerState{extLastSN: %d, extSecondLastSN: %d, extLastTS: %d, extSecondLastTS: %d)", r.ExtLastSN, r.ExtSecondLastSN, r.ExtLastTS, r.ExtSecondLastTS) + return fmt.Sprintf( + "RTPMungerState{extLastSN: %d, extSecondLastSN: %d, extLastTS: %d, extSecondLastTS: %d, lastMarker: %v, secondLastMarker: %v)", + r.ExtLastSN, r.ExtSecondLastSN, + r.ExtLastTS, r.ExtSecondLastTS, + r.LastMarker, r.SecondLastMarker, + ) } // ---------------------------------------------------------------------- @@ -77,7 +84,8 @@ type RTPMunger struct { extSecondLastTS uint64 tsOffset uint64 - lastMarker bool + lastMarker bool + secondLastMarker bool extRtxGateSn uint64 isInRtxGateRegion bool @@ -100,15 +108,18 @@ func (r *RTPMunger) DebugInfo() map[string]interface{} { "ExtSecondLastTS": r.extSecondLastTS, "TSOffset": r.tsOffset, "LastMarker": r.lastMarker, + "SecondLastMarker": r.secondLastMarker, } } func (r *RTPMunger) GetLast() RTPMungerState { return RTPMungerState{ - ExtLastSN: r.extLastSN, - ExtSecondLastSN: r.extSecondLastSN, - ExtLastTS: r.extLastTS, - ExtSecondLastTS: r.extSecondLastTS, + ExtLastSN: r.extLastSN, + ExtSecondLastSN: r.extSecondLastSN, + ExtLastTS: r.extLastTS, + ExtSecondLastTS: r.extSecondLastTS, + LastMarker: r.lastMarker, + SecondLastMarker: r.secondLastMarker, } } @@ -117,6 +128,8 @@ func (r *RTPMunger) SeedLast(state RTPMungerState) { r.extSecondLastSN = state.ExtSecondLastSN r.extLastTS = state.ExtLastTS r.extSecondLastTS = state.ExtSecondLastTS + r.lastMarker = state.LastMarker + r.secondLastMarker = state.SecondLastMarker } func (r *RTPMunger) SetLastSnTs(extPkt *buffer.ExtPacket) { @@ -164,9 +177,10 @@ func (r *RTPMunger) PacketDropped(extPkt *buffer.ExtPacket) { r.updateSnOffset() r.extLastTS = r.extSecondLastTS + r.lastMarker = r.secondLastMarker } -func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket) (*TranslationParamsRTP, error) { +func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket, marker bool) (*TranslationParamsRTP, error) { diff := int64(extPkt.ExtSequenceNumber - r.extHighestIncomingSN) if (diff == 1 && len(extPkt.Packet.Payload) != 0) || diff > 1 { // in-order - either contiguous packet with payload OR packet following a gap, may or may not have payload @@ -184,7 +198,8 @@ func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket) (*TranslationPara r.extLastSN = extMungedSN r.extSecondLastTS = r.extLastTS r.extLastTS = extMungedTS - r.lastMarker = extPkt.Packet.Marker + r.secondLastMarker = r.lastMarker + r.lastMarker = marker if extPkt.KeyFrame { r.extRtxGateSn = extMungedSN diff --git a/pkg/sfu/rtpmunger_test.go b/pkg/sfu/rtpmunger_test.go index 7c202ce1e..128d37b54 100644 --- a/pkg/sfu/rtpmunger_test.go +++ b/pkg/sfu/rtpmunger_test.go @@ -103,7 +103,7 @@ func TestPacketDropped(t *testing.T) { require.Equal(t, uint64(0), snOffset) require.Equal(t, uint64(0), r.tsOffset) - r.UpdateAndGetSnTs(extPkt) // update sequence number offset + r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) // update sequence number offset // drop a non-head packet, should cause no change in internals params = &testutils.TestExtPacketParams{ @@ -128,7 +128,7 @@ func TestPacketDropped(t *testing.T) { } extPkt, _ = testutils.GetTestExtPacket(params) - r.UpdateAndGetSnTs(extPkt) // update sequence number offset + r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) // update sequence number offset require.Equal(t, uint64(44444), r.extLastSN) r.PacketDropped(extPkt) @@ -147,7 +147,7 @@ func TestPacketDropped(t *testing.T) { } extPkt, _ = testutils.GetTestExtPacket(params) - r.UpdateAndGetSnTs(extPkt) // update sequence number offset + r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) // update sequence number offset require.Equal(t, r.extLastSN, uint64(44444)) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) @@ -165,7 +165,7 @@ func TestOutOfOrderSequenceNumber(t *testing.T) { } extPkt, _ := testutils.GetTestExtPacket(params) r.SetLastSnTs(extPkt) - r.UpdateAndGetSnTs(extPkt) + r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) // should not be able to add a missing sequence number to the cache that is before start err := r.snRangeMap.ExcludeRange(23332, 23333) @@ -180,7 +180,7 @@ func TestOutOfOrderSequenceNumber(t *testing.T) { } extPkt, _ = testutils.GetTestExtPacket(params) - tp, err := r.UpdateAndGetSnTs(extPkt) + tp, err := r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.Error(t, err) // add a missing sequence number to the cache @@ -194,7 +194,7 @@ func TestOutOfOrderSequenceNumber(t *testing.T) { PayloadSize: 10, } extPkt, _ = testutils.GetTestExtPacket(params) - r.UpdateAndGetSnTs(extPkt) + r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) // out-of-order sequence number should be munged from cache params = &testutils.TestExtPacketParams{ @@ -211,7 +211,7 @@ func TestOutOfOrderSequenceNumber(t *testing.T) { extTimestamp: 0xabcdef, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.Equal(t, tpExpected, *tp) @@ -227,7 +227,7 @@ func TestOutOfOrderSequenceNumber(t *testing.T) { snOrdering: SequenceNumberOrderingOutOfOrder, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.Error(t, err, ErrOutOfOrderSequenceNumberCacheMiss) require.Equal(t, tpExpected, *tp) } @@ -244,15 +244,14 @@ func TestDuplicateSequenceNumber(t *testing.T) { r.SetLastSnTs(extPkt) // send first packet through - r.UpdateAndGetSnTs(extPkt) + r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) // send it again - duplicate packet tpExpected := TranslationParamsRTP{ snOrdering: SequenceNumberOrderingDuplicate, } - tp, err := r.UpdateAndGetSnTs(extPkt) - require.Error(t, err) + tp, err := r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.ErrorIs(t, err, ErrDuplicatePacket) require.Equal(t, tpExpected, *tp) } @@ -273,7 +272,7 @@ func TestPaddingOnlyPacket(t *testing.T) { snOrdering: SequenceNumberOrderingContiguous, } - tp, err := r.UpdateAndGetSnTs(extPkt) + tp, err := r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.Error(t, err) require.ErrorIs(t, err, ErrPaddingOnlyPacket) require.Equal(t, tpExpected, *tp) @@ -296,7 +295,7 @@ func TestPaddingOnlyPacket(t *testing.T) { extTimestamp: 0xabcdef, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.Equal(t, tpExpected, *tp) require.Equal(t, uint64(23335), r.extHighestIncomingSN) @@ -318,7 +317,7 @@ func TestGapInSequenceNumber(t *testing.T) { extPkt, _ := testutils.GetTestExtPacket(params) r.SetLastSnTs(extPkt) - _, err := r.UpdateAndGetSnTs(extPkt) + _, err := r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) // three lost packets @@ -337,7 +336,7 @@ func TestGapInSequenceNumber(t *testing.T) { extTimestamp: 0xabcdef, } - tp, err := r.UpdateAndGetSnTs(extPkt) + tp, err := r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.Equal(t, tpExpected, *tp) require.Equal(t, uint64(65536+1), r.extHighestIncomingSN) @@ -366,7 +365,7 @@ func TestGapInSequenceNumber(t *testing.T) { snOrdering: SequenceNumberOrderingContiguous, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.ErrorIs(t, err, ErrPaddingOnlyPacket) require.Equal(t, tpExpected, *tp) require.Equal(t, uint64(65536+2), r.extHighestIncomingSN) @@ -390,7 +389,7 @@ func TestGapInSequenceNumber(t *testing.T) { extTimestamp: 0xabcdef, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.Equal(t, tpExpected, *tp) require.Equal(t, uint64(65536+4), r.extHighestIncomingSN) @@ -417,7 +416,7 @@ func TestGapInSequenceNumber(t *testing.T) { snOrdering: SequenceNumberOrderingContiguous, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.ErrorIs(t, err, ErrPaddingOnlyPacket) require.Equal(t, tpExpected, *tp) require.Equal(t, uint64(65536+5), r.extHighestIncomingSN) @@ -441,7 +440,7 @@ func TestGapInSequenceNumber(t *testing.T) { extTimestamp: 0xabcdef, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.Equal(t, tpExpected, *tp) require.Equal(t, uint64(65536+7), r.extHighestIncomingSN) @@ -474,7 +473,7 @@ func TestGapInSequenceNumber(t *testing.T) { extTimestamp: 0xabcdef, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.Equal(t, tpExpected, *tp) require.Equal(t, uint64(65536+7), r.extHighestIncomingSN) @@ -497,7 +496,7 @@ func TestGapInSequenceNumber(t *testing.T) { extTimestamp: 0xabcdef, } - tp, err = r.UpdateAndGetSnTs(extPkt) + tp, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.Equal(t, tpExpected, *tp) require.Equal(t, uint64(65536+7), r.extHighestIncomingSN) @@ -565,7 +564,7 @@ func TestIsOnFrameBoundary(t *testing.T) { r.SetLastSnTs(extPkt) // send it through - _, err := r.UpdateAndGetSnTs(extPkt) + _, err := r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.False(t, r.IsOnFrameBoundary()) @@ -580,7 +579,7 @@ func TestIsOnFrameBoundary(t *testing.T) { extPkt, _ = testutils.GetTestExtPacket(params) // send it through - _, err = r.UpdateAndGetSnTs(extPkt) + _, err = r.UpdateAndGetSnTs(extPkt, extPkt.Packet.Marker) require.NoError(t, err) require.True(t, r.IsOnFrameBoundary()) }