mirror of
https://github.com/livekit/livekit.git
synced 2026-10-22 19:43:01 +00:00
sfu: send padding-only packets with no payload (#4993)
This commit is contained in:
1 parent
93296ad4f6
commit
5dac0aab20
4 files changed
+39
-38
No files matched your search
+8
-23
@@ -86,7 +86,7 @@ var (
|
||||
)
|
||||
|
||||
// padding of a dropped packet's slot, one byte: the count byte itself
|
||||
var droppedSlotPadding = []byte{1}
|
||||
const droppedSlotPaddingSize = 1
|
||||
|
||||
var (
|
||||
VP8KeyFrame8x8 = []byte{
|
||||
@@ -1253,7 +1253,7 @@ func (d *DownTrack) writeDroppedSlotPadding(extSequenceNumber uint64, extTimesta
|
||||
initPooledRTPHeader(hdr, rtp.Header{
|
||||
Version: 2,
|
||||
Padding: true,
|
||||
PaddingSize: byte(len(droppedSlotPadding)),
|
||||
PaddingSize: droppedSlotPaddingSize,
|
||||
Marker: false,
|
||||
PayloadType: uint8(d.payloadType.Load()),
|
||||
SequenceNumber: uint16(extSequenceNumber),
|
||||
@@ -1270,7 +1270,7 @@ func (d *DownTrack) writeDroppedSlotPadding(extSequenceNumber uint64, extTimesta
|
||||
hdr.Marker,
|
||||
hdrSize,
|
||||
0,
|
||||
len(droppedSlotPadding),
|
||||
droppedSlotPaddingSize,
|
||||
isOutOfOrder,
|
||||
)
|
||||
|
||||
@@ -1279,7 +1279,6 @@ func (d *DownTrack) writeDroppedSlotPadding(extSequenceNumber uint64, extTimesta
|
||||
Header: hdr,
|
||||
HeaderPool: RTPHeaderFactory,
|
||||
HeaderSize: hdrSize,
|
||||
Payload: droppedSlotPadding,
|
||||
ProbeClusterId: ccutils.ProbeClusterIdInvalid, // not probe traffic
|
||||
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
|
||||
TransportWideExtID: uint8(d.transportWideExtID),
|
||||
@@ -1354,7 +1353,6 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
|
||||
}
|
||||
|
||||
bytesSent := 0
|
||||
payloads := make([]byte, RTPPaddingMaxPayloadSize*len(snts))
|
||||
for i := range snts {
|
||||
hdr := RTPHeaderFactory.Get().(*rtp.Header)
|
||||
initPooledRTPHeader(hdr, rtp.Header{
|
||||
@@ -1369,12 +1367,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
|
||||
})
|
||||
d.addDummyExtensions(hdr)
|
||||
|
||||
payload := payloads[i*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize]
|
||||
// last byte of padding has padding size including that byte
|
||||
payload[RTPPaddingMaxPayloadSize-1] = byte(RTPPaddingMaxPayloadSize)
|
||||
|
||||
hdrSize := hdr.MarshalSize()
|
||||
payloadSize := len(payload)
|
||||
d.rtpStats.Update(
|
||||
mono.UnixNano(),
|
||||
snts[i].extSequenceNumber,
|
||||
@@ -1382,7 +1375,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
|
||||
hdr.Marker,
|
||||
hdrSize,
|
||||
0,
|
||||
payloadSize,
|
||||
RTPPaddingMaxPayloadSize,
|
||||
false,
|
||||
)
|
||||
|
||||
@@ -1391,7 +1384,6 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
|
||||
Header: hdr,
|
||||
HeaderPool: RTPHeaderFactory,
|
||||
HeaderSize: hdrSize,
|
||||
Payload: payload,
|
||||
ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()),
|
||||
IsProbe: true,
|
||||
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
|
||||
@@ -1400,7 +1392,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
|
||||
}
|
||||
d.pacer.Enqueue(pacerPacket)
|
||||
|
||||
bytesSent += hdrSize + payloadSize
|
||||
bytesSent += hdrSize + RTPPaddingMaxPayloadSize
|
||||
}
|
||||
|
||||
return bytesSent
|
||||
@@ -2528,8 +2520,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int {
|
||||
return 0
|
||||
}
|
||||
|
||||
payloads := make([]byte, RTPPaddingMaxPayloadSize*num)
|
||||
for i := range num {
|
||||
for range num {
|
||||
rtxExtSequenceNumber := d.rtxSequenceNumber.Inc()
|
||||
hdr := RTPHeaderFactory.Get().(*rtp.Header)
|
||||
initPooledRTPHeader(hdr, rtp.Header{
|
||||
@@ -2544,12 +2535,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int {
|
||||
})
|
||||
d.addDummyExtensions(hdr)
|
||||
|
||||
payload := payloads[i*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize]
|
||||
// last byte of padding has padding size including that byte
|
||||
payload[RTPPaddingMaxPayloadSize-1] = byte(RTPPaddingMaxPayloadSize)
|
||||
|
||||
hdrSize := hdr.MarshalSize()
|
||||
payloadSize := len(payload)
|
||||
d.rtpStatsRTX.Update(
|
||||
mono.UnixNano(),
|
||||
rtxExtSequenceNumber,
|
||||
@@ -2557,7 +2543,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int {
|
||||
hdr.Marker,
|
||||
hdrSize,
|
||||
0,
|
||||
payloadSize,
|
||||
RTPPaddingMaxPayloadSize,
|
||||
false,
|
||||
)
|
||||
pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet)
|
||||
@@ -2565,7 +2551,6 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int {
|
||||
Header: hdr,
|
||||
HeaderPool: RTPHeaderFactory,
|
||||
HeaderSize: hdrSize,
|
||||
Payload: payload,
|
||||
ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()),
|
||||
IsProbe: true,
|
||||
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
|
||||
@@ -2574,7 +2559,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int {
|
||||
}
|
||||
d.pacer.Enqueue(pacerPacket)
|
||||
|
||||
bytesSent += hdrSize + payloadSize
|
||||
bytesSent += hdrSize + RTPPaddingMaxPayloadSize
|
||||
}
|
||||
} else {
|
||||
src := PacketFactory.Get().(*[]byte)
|
||||
|
||||
@@ -584,10 +584,10 @@ func lktsTrailer() []byte {
|
||||
// the primary SSRC/payload type, carries the padding bit, and declares a full
|
||||
// padding block.
|
||||
//
|
||||
// A padding-only packet is entirely padding; pion/srtp serializes the padding block
|
||||
// from header.PaddingSize (via rtp.MarshalPacketTo), and the pion receiver strips it
|
||||
// and reports both the padding bit and the padding size, so the received payload
|
||||
// length equals the declared padding size (RTPPaddingMaxPayloadSize).
|
||||
// A padding-only packet is entirely padding: pion/srtp serializes the padding block
|
||||
// from header.PaddingSize (via rtp.MarshalPacketTo) after the payload, so the packet
|
||||
// has no payload. A receiver takes a packet with any payload left after the padding
|
||||
// as media.
|
||||
func TestDownTrackSendsPaddingOnlyPackets(t *testing.T) {
|
||||
h := vnettest.NewHosts(t)
|
||||
factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory()
|
||||
@@ -665,12 +665,27 @@ func TestDownTrackSendsPaddingOnlyPackets(t *testing.T) {
|
||||
// a padding-only packet: it must declare a full padding block.
|
||||
require.EqualValues(t, RTPPaddingMaxPayloadSize, pkt.Header.PaddingSize,
|
||||
"padding-only packet must declare full padding size (sn=%d)", pkt.SequenceNumber)
|
||||
require.Equal(t, RTPPaddingMaxPayloadSize, len(pkt.Payload),
|
||||
"padding-only packet must serialize a full padding block (sn=%d)", pkt.SequenceNumber)
|
||||
require.Empty(t, pkt.Payload, "padding-only packet must have no payload (sn=%d)", pkt.SequenceNumber)
|
||||
sawPadding++
|
||||
}
|
||||
require.Equal(t, 1, sawKeyframe, "expected exactly one forwarded media keyframe")
|
||||
require.GreaterOrEqual(t, sawPadding, numPadding, "expected the padding-only packets to be observed")
|
||||
|
||||
// the padding in the slot of a dropped packet has no payload either
|
||||
dh.dt.WriteDroppedSlotPaddingForTest(6000, 190000)
|
||||
var pkt *rtp.Packet
|
||||
require.Eventually(t, func() bool {
|
||||
for _, p := range dh.capture.all() {
|
||||
if p.SequenceNumber == 6000 {
|
||||
pkt = p
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}, 10*time.Second, 20*time.Millisecond, "expected the dropped slot padding to arrive on the far side")
|
||||
require.True(t, pkt.Header.Padding)
|
||||
require.EqualValues(t, DroppedSlotPaddingSizeForTest, pkt.Header.PaddingSize)
|
||||
require.Empty(t, pkt.Payload)
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
@@ -681,7 +696,6 @@ func TestDownTrackSendsPaddingOnlyPackets(t *testing.T) {
|
||||
type capturingProbe struct {
|
||||
hdr rtp.Header
|
||||
payloadLen int
|
||||
lastByte byte
|
||||
}
|
||||
|
||||
// capturedRTX records what the DownTrack handed the pacer for an RTX packet.
|
||||
@@ -706,9 +720,6 @@ func (c *capturingPacer) Enqueue(p *pacer.Packet) {
|
||||
switch {
|
||||
case p.IsProbe:
|
||||
cp := capturingProbe{hdr: *p.Header, payloadLen: len(p.Payload)}
|
||||
if len(p.Payload) > 0 {
|
||||
cp.lastByte = p.Payload[len(p.Payload)-1]
|
||||
}
|
||||
c.lock.Lock()
|
||||
c.probes = append(c.probes, cp)
|
||||
c.lock.Unlock()
|
||||
@@ -900,16 +911,15 @@ func TestDownTrackSendsProbePackets(t *testing.T) {
|
||||
}, 5*time.Second, 20*time.Millisecond, "expected WriteProbePackets to emit RTX probe packets")
|
||||
|
||||
// every probe packet must be on the RTX stream, carry the padding bit, declare
|
||||
// the full padding size, and serialize a full padding block ending in the
|
||||
// padding count byte.
|
||||
// the full padding size, and have no payload: pion serializes the padding from
|
||||
// the header.
|
||||
for _, pr := range cp.probePackets() {
|
||||
require.EqualValues(t, 2, pr.hdr.Version, "probe packet version")
|
||||
require.EqualValues(t, dh.dt.SSRCRTX(), pr.hdr.SSRC, "probe packet must be on the RTX stream")
|
||||
require.EqualValues(t, dh.dt.PayloadTypeRTXForTest(), pr.hdr.PayloadType, "probe packet must use the RTX payload type")
|
||||
require.True(t, pr.hdr.Padding, "probe packet must carry padding bit")
|
||||
require.EqualValues(t, RTPPaddingMaxPayloadSize, pr.hdr.PaddingSize, "probe packet must set header.PaddingSize")
|
||||
require.Equal(t, RTPPaddingMaxPayloadSize, pr.payloadLen, "probe packet must carry a full padding block")
|
||||
require.EqualValues(t, RTPPaddingMaxPayloadSize, pr.lastByte, "probe packet trailing byte must be the padding count")
|
||||
require.Zero(t, pr.payloadLen, "probe packet must have no payload")
|
||||
}
|
||||
|
||||
// the BWE is wired and exercised (the pacer records every send into it);
|
||||
|
||||
@@ -32,6 +32,12 @@ func (d *DownTrack) PayloadTypeRTXForTest() uint32 { return d.payloadTypeRTX.Loa
|
||||
|
||||
func (d *DownTrack) RTPStatsActiveForTest() bool { return d.rtpStats.IsActive() }
|
||||
|
||||
const DroppedSlotPaddingSizeForTest = droppedSlotPaddingSize
|
||||
|
||||
func (d *DownTrack) WriteDroppedSlotPaddingForTest(extSequenceNumber uint64, extTimestamp uint64) {
|
||||
d.writeDroppedSlotPadding(extSequenceNumber, extTimestamp, false)
|
||||
}
|
||||
|
||||
func (d *DownTrack) LastReceiverReportTimeForTest() int64 {
|
||||
return d.rtpStats.LastReceiverReportTime()
|
||||
}
|
||||
|
||||
@@ -110,7 +110,7 @@ func (b *Base) patchRTPHeaderExtensions(p *Packet) error {
|
||||
b.lastPacketSentAt.Store(sendingAt.UnixNano())
|
||||
}
|
||||
|
||||
packetSize := p.HeaderSize + len(p.Payload)
|
||||
packetSize := p.HeaderSize + len(p.Payload) + int(p.Header.PaddingSize)
|
||||
if p.TransportWideExtID != 0 && b.bwe != nil {
|
||||
twccSN := b.bwe.RecordPacketSendAndGetSequenceNumber(
|
||||
sendingAt.UnixMicro(),
|
||||
|
||||
Reference in new issue
Block a user