diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index e01c96748..8c0d87977 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -85,6 +85,9 @@ var ( errPayloadOverflow = errors.New("payload overflow") ) +// padding of a dropped packet's slot, one byte: the count byte itself +var droppedSlotPadding = []byte{1} + var ( VP8KeyFrame8x8 = []byte{ 0x10, 0x02, 0x00, 0x9d, 0x01, 0x2a, 0x08, 0x00, @@ -1032,6 +1035,9 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { if err != nil { d.params.Logger.Errorw("could not get translation params", err) } + if tp.padDroppedSlot { + d.writeDroppedSlotPadding(tp.rtp.extSequenceNumber, tp.rtp.extTimestamp) + } return 0 } @@ -1226,6 +1232,50 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { return 1 } +// writeDroppedSlotPadding sends a padding only packet in the outgoing slot of a dropped +// out-of-order packet. The receiver takes it as padding for that sequence number and +// does not NACK it. The sequencer has no entry for the slot, so a NACK that still comes +// finds nothing to retransmit. +func (d *DownTrack) writeDroppedSlotPadding(extSequenceNumber uint64, extTimestamp uint64) { + hdr := RTPHeaderFactory.Get().(*rtp.Header) + initPooledRTPHeader(hdr, rtp.Header{ + Version: 2, + Padding: true, + PaddingSize: byte(len(droppedSlotPadding)), + Marker: false, + PayloadType: uint8(d.payloadType.Load()), + SequenceNumber: uint16(extSequenceNumber), + Timestamp: uint32(extTimestamp), + SSRC: d.ssrc, + }) + d.addDummyExtensions(hdr) + + hdrSize := hdr.MarshalSize() + d.rtpStats.Update( + mono.UnixNano(), + extSequenceNumber, + extTimestamp, + hdr.Marker, + hdrSize, + 0, + len(droppedSlotPadding), + true, + ) + + pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet) + *pacerPacket = pacer.Packet{ + Header: hdr, + HeaderPool: RTPHeaderFactory, + HeaderSize: hdrSize, + Payload: droppedSlotPadding, + ProbeClusterId: ccutils.ProbeClusterIdInvalid, // not probe traffic + AbsSendTimeExtID: uint8(d.absSendTimeExtID), + TransportWideExtID: uint8(d.transportWideExtID), + WriteStream: d.writeStream, + } + d.pacer.Enqueue(pacerPacket) +} + // WritePaddingRTP tries to write as many padding only RTP packets as necessary // to satisfy given size to the DownTrack func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMarker bool) int { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 3a2e8359c..c546796fa 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -209,6 +209,11 @@ type TranslationParams struct { marker bool // end of the svc spatial layer frame isEndOfLayerFrame bool + // The packet is dropped, but the RTP munger gave its outgoing sequence number + // away when it saw the gap, so the downtrack sends padding in that slot. Without + // it the subscriber NACKs a sequence number nothing can answer, and a VP8 + // receiver without picture IDs stalls on the hole until a key frame. + padDroppedSlot bool } // codecHeader returns the munged codec header to prepend to the payload. @@ -2226,8 +2231,12 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in if f.started && result.IsRelevant { // call to update highest incoming sequence number and other internal structures if tpRTP, err := f.rtpMunger.UpdateAndGetSnTs(extPkt, result.RTPMarker); err == nil { - if tpRTP.snOrdering == SequenceNumberOrderingContiguous { + switch tpRTP.snOrdering { + case SequenceNumberOrderingContiguous: f.rtpMunger.PacketDropped(extPkt) + case SequenceNumberOrderingOutOfOrder: + tp.rtp = tpRTP + tp.padDroppedSlot = true } } } @@ -2287,6 +2296,9 @@ func (f *Forwarder) translateCodecHeader(extPkt *buffer.ExtPacket, tp *Translati // filtered temporal layer, update sequence number offset to prevent holes f.rtpMunger.PacketDropped(extPkt) } + if tp.rtp.snOrdering == SequenceNumberOrderingOutOfOrder { + tp.padDroppedSlot = true + } return nil } diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index ef657004b..cc7f92f0d 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -2113,6 +2113,56 @@ func TestForwarderVP8DDWithPictureID(t *testing.T) { require.True(t, f.isDDAvailable, "temporal layers still switch with the dependency descriptor") } +func TestForwarderPadDroppedSlot(t *testing.T) { + f := NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) + f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) + f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + + packet := func(sn uint16, frameNum uint64, tid int, isOutOfOrder bool) *buffer.ExtPacket { + extPkt, err := testutils.GetTestExtPacketVP8DD( + &testutils.TestExtPacketParams{SequenceNumber: sn, Timestamp: 0xabcdef, SSRC: 0x12345678, PayloadSize: 20, Marker: true, IsOutOfOrder: isOutOfOrder}, + &testutils.TestDDParams{ + FirstPacketInFrame: true, + ExtFrameNum: frameNum, + TemporalID: tid, + DTIs: "SSS", + DecodeTargets: testutils.TestVP8DecodeTargets, + }, + ) + require.NoError(t, err) + return extPkt + } + + tp, err := f.GetTranslationParams(packet(10, 0, 0, false), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + + // a gap, the munger hands out a slot for sequence number 11 + tp, err = f.GetTranslationParams(packet(12, 2, 0, false), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.Equal(t, uint64(12), tp.rtp.extSequenceNumber) + + // the late packet is on a filtered layer, its slot gets padding + tp, err = f.GetTranslationParams(packet(11, 1, 2, true), 0) + require.NoError(t, err) + require.True(t, tp.shouldDrop) + require.True(t, tp.padDroppedSlot) + require.Equal(t, uint64(11), tp.rtp.extSequenceNumber) + require.Equal(t, uint64(0xabcdef), tp.rtp.extTimestamp) + + // an in-order drop rolls the slot back instead + tp, err = f.GetTranslationParams(packet(13, 3, 2, false), 0) + require.NoError(t, err) + require.True(t, tp.shouldDrop) + require.False(t, tp.padDroppedSlot) + tp, err = f.GetTranslationParams(packet(14, 4, 0, false), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.Equal(t, uint64(13), tp.rtp.extSequenceNumber) +} + func TestForwarderGetSnTsForPadding(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo)