mirror of
https://github.com/livekit/livekit.git
synced 2026-10-15 13:20:13 +00:00
sfu: send padding in the slot of a dropped out-of-order packet (#4979)
On a gap the RTP munger gives the missing sequence numbers outgoing slots. If a late packet then arrives and is dropped, nothing is ever sent in its slot, and a subscriber that finds frame references by sequence number stalls until a key frame. The forwarder now flags such drops, and the downtrack sends a one byte padding only packet in the slot. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
152328c76a
commit
a67eb7b560
@@ -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 {
|
||||
|
||||
+13
-1
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user