dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 02:54:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Log PLI requests. (#2194)

Browse Source 
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
This commit is contained in:
Raja Subramanian
2023-10-26 21:07:36 +05:30
committed by GitHub
parent ce8f64176a
commit 8b16db2270
5 changed files with 59 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/downtrack.go
+7 -5
View File
@@ -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
pkg/sfu/forwarder.go
+2 -2
View File
@@ -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)
}
}
pkg/sfu/receiver.go
+1 -1
View File
@@ -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)
})
})
pkg/sfu/rtpmunger.go
+27 -12
View File
@@ -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
pkg/sfu/rtpmunger_test.go
+22 -23
View File
@@ -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())
}