mirror of
https://github.com/livekit/livekit.git
synced 2026-10-23 15:12:43 +00:00
Merge branch 'dc/feature/fec_up' of github.com:livekit/livekit into dc/feature/fec_up
This commit is contained in:
David Chen committed
commit
582eaa96b3
6 files changed
+224
-53
No files matched your search
@@ -452,7 +452,7 @@ func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) {
|
||||
}
|
||||
}
|
||||
|
||||
func (b *Buffer) markAsFECRepair() {
|
||||
func (b *Buffer) setFECRepairBufferLimit() {
|
||||
b.Lock()
|
||||
defer b.Unlock()
|
||||
|
||||
@@ -564,6 +564,7 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) {
|
||||
b.fecSSRC = fecPkt.SSRC
|
||||
b.maybeCreateFECDecoderLocked()
|
||||
if b.fecDecoder == nil {
|
||||
b.logger.Infow("flexfec decoder unavailable after creation attempt", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC(), "payloadType", b.fecPayloadType)
|
||||
b.Unlock()
|
||||
return
|
||||
}
|
||||
@@ -582,6 +583,20 @@ func (b *Buffer) feedFECLocked(
|
||||
) (fecRecoveryDelta, func(received int, recovered int, discarded int, bytesReceived int)) {
|
||||
statsBefore := b.fecDecoder.Stats()
|
||||
recovered := b.fecDecoder.DecodeFEC(pkt)
|
||||
statsAfter := b.fecDecoder.Stats()
|
||||
|
||||
fecPackets := statsAfter.FECPacketsReceived - statsBefore.FECPacketsReceived
|
||||
fecBytes := statsAfter.FECBytesReceived - statsBefore.FECBytesReceived
|
||||
fecPacketsDiscarded := statsAfter.FECPacketsDiscarded - statsBefore.FECPacketsDiscarded
|
||||
fecPacketsRecovered := statsAfter.PacketsRecovered - statsBefore.PacketsRecovered
|
||||
if b.rtpStats != nil {
|
||||
b.rtpStats.UpdateFEC(
|
||||
fecPackets,
|
||||
fecBytes,
|
||||
fecPacketsDiscarded,
|
||||
fecPacketsRecovered,
|
||||
)
|
||||
}
|
||||
|
||||
if len(recovered) > 0 && b.fecPktBuf == nil {
|
||||
b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize)
|
||||
@@ -602,12 +617,11 @@ func (b *Buffer) feedFECLocked(
|
||||
}
|
||||
|
||||
if cb := b.onFECRecovery; cb != nil {
|
||||
statsAfter := b.fecDecoder.Stats()
|
||||
return fecRecoveryDelta{
|
||||
received: int(statsAfter.FECPacketsReceived - statsBefore.FECPacketsReceived),
|
||||
recovered: len(recovered),
|
||||
discarded: int(statsAfter.FECPacketsDiscarded - statsBefore.FECPacketsDiscarded),
|
||||
bytesReceived: int(statsAfter.FECBytesReceived - statsBefore.FECBytesReceived),
|
||||
received: int(fecPackets),
|
||||
recovered: int(fecPacketsRecovered),
|
||||
discarded: int(fecPacketsDiscarded),
|
||||
bytesReceived: int(fecBytes),
|
||||
}, cb
|
||||
}
|
||||
|
||||
|
||||
@@ -18,7 +18,6 @@ import (
|
||||
"crypto/aes"
|
||||
"crypto/cipher"
|
||||
"encoding/binary"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -31,6 +30,7 @@ import (
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/sfu/testutils/rtptest"
|
||||
"github.com/livekit/mediatransportutil/pkg/twcc"
|
||||
)
|
||||
|
||||
@@ -49,34 +49,6 @@ var flexFECCodec = webrtc.RTPCodecParameters{
|
||||
PayloadType: webrtc.PayloadType(fecTestFECPT),
|
||||
}
|
||||
|
||||
func fecTestMediaPackets(t *testing.T, baseSN uint16, count int) []rtp.Packet {
|
||||
t.Helper()
|
||||
rng := rand.New(rand.NewSource(int64(baseSN)))
|
||||
pkts := make([]rtp.Packet, 0, count)
|
||||
for i := 0; i < count; i++ {
|
||||
payload := make([]byte, 50+rng.Intn(200))
|
||||
rng.Read(payload)
|
||||
// valid VP8 payload descriptor (S=1, no extensions) so the video
|
||||
// packet processing in the buffer accepts the packet
|
||||
payload[0] = 0x10
|
||||
sn := baseSN + uint16(i)
|
||||
pkts = append(pkts, rtp.Packet{
|
||||
Header: rtp.Header{
|
||||
Version: 2,
|
||||
PayloadType: uint8(vp8Codec.PayloadType),
|
||||
SequenceNumber: sn,
|
||||
// derive timestamp from the sequence number so windows
|
||||
// generated separately stay monotonic
|
||||
Timestamp: 90000 + 3000*uint32(sn),
|
||||
SSRC: fecTestMediaSSRC,
|
||||
Marker: i == count-1,
|
||||
},
|
||||
Payload: payload,
|
||||
})
|
||||
}
|
||||
return pkts
|
||||
}
|
||||
|
||||
func bindFECTestBuffer(t *testing.T, buff *Buffer) {
|
||||
t.Helper()
|
||||
buff.codecType = webrtc.RTPCodecTypeVideo
|
||||
@@ -195,7 +167,7 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) {
|
||||
receivedDelta += received
|
||||
})
|
||||
|
||||
media := fecTestMediaPackets(t, 100, 10)
|
||||
media := rtptest.GenerateVP8Packets(100, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC)
|
||||
fecPackets := encoder.EncodeFec(media, 2)
|
||||
require.NotEmpty(t, fecPackets)
|
||||
@@ -218,6 +190,17 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) {
|
||||
assert.Equal(t, 1, recoveredDelta)
|
||||
assert.Equal(t, len(fecPackets), receivedDelta)
|
||||
|
||||
fecPayloadBytes := 0
|
||||
for i := range fecPackets {
|
||||
fecPayloadBytes += len(fecPackets[i].Payload)
|
||||
}
|
||||
rtpStats := primary.GetStats()
|
||||
require.NotNil(t, rtpStats)
|
||||
assert.EqualValues(t, len(fecPackets), rtpStats.FecPackets)
|
||||
assert.EqualValues(t, fecPayloadBytes, rtpStats.FecBytes)
|
||||
assert.EqualValues(t, 0, rtpStats.FecPacketsDiscarded)
|
||||
assert.EqualValues(t, 1, rtpStats.FecPacketsRecovered)
|
||||
|
||||
// the 9 received packets flow through the ext packet pipeline, the
|
||||
// recovered one fills the bucket like an RTX repair
|
||||
extSNBySN := readExtSequenceNumbers(t, primary, len(media)-1)
|
||||
@@ -237,7 +220,7 @@ func TestBufferFECRecoversEncryptedPayload(t *testing.T) {
|
||||
aead, err := cipher.NewGCM(block)
|
||||
require.NoError(t, err)
|
||||
|
||||
media := fecTestMediaPackets(t, 120, 5)
|
||||
media := rtptest.GenerateVP8Packets(120, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
plaintexts := make([][]byte, len(media))
|
||||
for i := range media {
|
||||
plaintexts[i] = append([]byte("encrypted-video-frame-"), byte(i))
|
||||
@@ -293,7 +276,7 @@ func TestBufferFECRecoveryCallbackCanReenterBuffer(t *testing.T) {
|
||||
callbackDone <- struct{}{}
|
||||
})
|
||||
|
||||
media := fecTestMediaPackets(t, 150, 5)
|
||||
media := rtptest.GenerateVP8Packets(150, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
for i := range media {
|
||||
writePacket(t, primary, &media[i])
|
||||
}
|
||||
@@ -332,7 +315,7 @@ func TestBufferFECPairAfterPackets(t *testing.T) {
|
||||
bindFECTestBuffer(t, primary)
|
||||
|
||||
encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC)
|
||||
media := fecTestMediaPackets(t, 200, 10)
|
||||
media := rtptest.GenerateVP8Packets(200, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
fecPackets := encoder.EncodeFec(media, 2)
|
||||
require.NotEmpty(t, fecPackets)
|
||||
|
||||
@@ -361,7 +344,7 @@ func TestBufferFECPairAfterPackets(t *testing.T) {
|
||||
assert.EqualValues(t, 1, stats.PacketsRecovered)
|
||||
|
||||
// the next window recovers normally
|
||||
media2 := fecTestMediaPackets(t, 210, 10)
|
||||
media2 := rtptest.GenerateVP8Packets(210, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
fecPackets2 := encoder.EncodeFec(media2, 2)
|
||||
require.NotEmpty(t, fecPackets2)
|
||||
|
||||
@@ -393,7 +376,7 @@ func TestBufferFECCoupledBeforeBuffersExist(t *testing.T) {
|
||||
fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer)
|
||||
bindFECTestBuffer(t, primary)
|
||||
|
||||
media := fecTestMediaPackets(t, 300, 5)
|
||||
media := rtptest.GenerateVP8Packets(300, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC)
|
||||
fecPackets := encoder.EncodeFec(media, 1)
|
||||
require.NotEmpty(t, fecPackets)
|
||||
@@ -424,7 +407,7 @@ func TestBufferFECSequenceNumberWrap(t *testing.T) {
|
||||
fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer)
|
||||
bindFECTestBuffer(t, primary)
|
||||
|
||||
media := fecTestMediaPackets(t, 65533, 5)
|
||||
media := rtptest.GenerateVP8Packets(65533, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
for i := range media {
|
||||
media[i].Timestamp = 90000 + 3000*uint32(i)
|
||||
}
|
||||
@@ -454,7 +437,7 @@ func TestBufferFECRecoveryAfterPaddingRemoval(t *testing.T) {
|
||||
fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer)
|
||||
bindFECTestBuffer(t, primary)
|
||||
|
||||
media := fecTestMediaPackets(t, 800, 5)
|
||||
media := rtptest.GenerateVP8Packets(800, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1)
|
||||
require.Len(t, fecPackets, 1)
|
||||
|
||||
@@ -523,7 +506,7 @@ func TestBufferFECIgnoresUnexpectedPayloadType(t *testing.T) {
|
||||
Codecs: []webrtc.RTPCodecParameters{vp8Codec},
|
||||
}, vp8Codec.RTPCodecCapability, 0))
|
||||
|
||||
media := fecTestMediaPackets(t, 400, 5)
|
||||
media := rtptest.GenerateVP8Packets(400, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC)
|
||||
fecPackets := encoder.EncodeFec(media, 1)
|
||||
require.NotEmpty(t, fecPackets)
|
||||
@@ -576,7 +559,7 @@ func TestBufferFECNACKSuppression(t *testing.T) {
|
||||
factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC)
|
||||
bindFECTestBuffer(t, primary)
|
||||
|
||||
media := fecTestMediaPackets(t, 700, 10)
|
||||
media := rtptest.GenerateVP8Packets(700, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
|
||||
encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC)
|
||||
fecPackets := encoder.EncodeFec(media, 2)
|
||||
require.NotEmpty(t, fecPackets)
|
||||
@@ -631,7 +614,7 @@ func TestBufferFECNACKNotSentForRecoveredPacket(t *testing.T) {
|
||||
|
||||
// single FEC packet protecting the whole window; dropping two of the
|
||||
// protected packets keeps the window unrecoverable until one arrives
|
||||
media := fecTestMediaPackets(t, 100, 5) // 100..104
|
||||
media := rtptest.GenerateVP8Packets(100, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) // 100..104
|
||||
fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1)
|
||||
require.Len(t, fecPackets, 1)
|
||||
|
||||
@@ -640,9 +623,9 @@ func TestBufferFECNACKNotSentForRecoveredPacket(t *testing.T) {
|
||||
writePacket(t, primary, &media[4]) // 104 -> gap 102, 103
|
||||
|
||||
// an unrelated gap outside the FEC-protected window that FEC cannot recover
|
||||
extra := fecTestMediaPackets(t, 105, 3) // 105..107
|
||||
writePacket(t, primary, &extra[0]) // 105
|
||||
writePacket(t, primary, &extra[2]) // 107 -> gap 106
|
||||
extra := rtptest.GenerateVP8Packets(105, 3, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) // 105..107
|
||||
writePacket(t, primary, &extra[0]) // 105
|
||||
writePacket(t, primary, &extra[2]) // 107 -> gap 106
|
||||
const unrecoverableSN = uint16(106)
|
||||
|
||||
// FEC packet retained: two protected packets (102, 103) still missing
|
||||
|
||||
@@ -93,7 +93,7 @@ func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io
|
||||
}
|
||||
for fec, base := range f.fecPair {
|
||||
if fec == ssrc {
|
||||
buffer.markAsFECRepair()
|
||||
buffer.setFECRepairBufferLimit()
|
||||
baseBuffer, ok := f.rtpBuffers[base]
|
||||
if ok {
|
||||
buffer.SetPrimaryBufferForFEC(baseBuffer)
|
||||
@@ -175,7 +175,7 @@ func (f *Factory) SetFECPair(fec, base uint32) {
|
||||
}
|
||||
f.Unlock()
|
||||
if fecBuffer != nil && baseBuffer == nil {
|
||||
fecBuffer.markAsFECRepair()
|
||||
fecBuffer.setFECRepairBufferLimit()
|
||||
}
|
||||
if fecBuffer != nil && baseBuffer != nil {
|
||||
fecBuffer.SetPrimaryBufferForFEC(baseBuffer)
|
||||
|
||||
@@ -48,6 +48,10 @@ type RTPDeltaInfo struct {
|
||||
PacketsPadding uint32
|
||||
BytesPadding uint64
|
||||
HeaderBytesPadding uint64
|
||||
FecPackets uint32
|
||||
FecBytes uint64
|
||||
FecPacketsDiscarded uint32
|
||||
FecPacketsRecovered uint32
|
||||
PacketsLost uint32
|
||||
PacketsMissing uint32
|
||||
PacketsOutOfOrder uint32
|
||||
@@ -76,6 +80,10 @@ func (r *RTPDeltaInfo) MarshalLogObject(e zapcore.ObjectEncoder) error {
|
||||
e.AddUint32("PacketsPadding", r.PacketsPadding)
|
||||
e.AddUint64("BytesPadding", r.BytesPadding)
|
||||
e.AddUint64("HeaderBytesPadding", r.HeaderBytesPadding)
|
||||
e.AddUint32("FecPackets", r.FecPackets)
|
||||
e.AddUint64("FecBytes", r.FecBytes)
|
||||
e.AddUint32("FecPacketsDiscarded", r.FecPacketsDiscarded)
|
||||
e.AddUint32("FecPacketsRecovered", r.FecPacketsRecovered)
|
||||
e.AddUint32("PacketsLost", r.PacketsLost)
|
||||
e.AddUint32("PacketsMissing", r.PacketsMissing)
|
||||
e.AddUint32("PacketsOutOfOrder", r.PacketsOutOfOrder)
|
||||
@@ -104,6 +112,11 @@ type snapshot struct {
|
||||
bytesPadding uint64
|
||||
headerBytesPadding uint64
|
||||
|
||||
fecPackets uint64
|
||||
fecBytes uint64
|
||||
fecPacketsDiscarded uint64
|
||||
fecPacketsRecovered uint64
|
||||
|
||||
frames uint32
|
||||
|
||||
plis uint32
|
||||
@@ -126,6 +139,10 @@ func (s *snapshot) MarshalLogObject(e zapcore.ObjectEncoder) error {
|
||||
e.AddUint64("packetsPadding", s.packetsPadding)
|
||||
e.AddUint64("bytesPadding", s.bytesPadding)
|
||||
e.AddUint64("headerBytesPadding", s.headerBytesPadding)
|
||||
e.AddUint64("fecPackets", s.fecPackets)
|
||||
e.AddUint64("fecBytes", s.fecBytes)
|
||||
e.AddUint64("fecPacketsDiscarded", s.fecPacketsDiscarded)
|
||||
e.AddUint64("fecPacketsRecovered", s.fecPacketsRecovered)
|
||||
e.AddUint32("frames", s.frames)
|
||||
e.AddUint32("plis", s.plis)
|
||||
e.AddUint32("firs", s.firs)
|
||||
@@ -222,6 +239,11 @@ type rtpStatsBase struct {
|
||||
bytesPadding uint64
|
||||
headerBytesPadding uint64
|
||||
|
||||
fecPackets uint64
|
||||
fecBytes uint64
|
||||
fecPacketsDiscarded uint64
|
||||
fecPacketsRecovered uint64
|
||||
|
||||
frames uint32
|
||||
|
||||
jitter float64
|
||||
@@ -285,6 +307,11 @@ func (r *rtpStatsBase) seed(from *rtpStatsBase) bool {
|
||||
r.bytesPadding = from.bytesPadding
|
||||
r.headerBytesPadding = from.headerBytesPadding
|
||||
|
||||
r.fecPackets = from.fecPackets
|
||||
r.fecBytes = from.fecBytes
|
||||
r.fecPacketsDiscarded = from.fecPacketsDiscarded
|
||||
r.fecPacketsRecovered = from.fecPacketsRecovered
|
||||
|
||||
r.frames = from.frames
|
||||
|
||||
r.jitter = from.jitter
|
||||
@@ -324,6 +351,25 @@ func (r *rtpStatsBase) newSnapshotID(extStartSN uint64) uint32 {
|
||||
return id
|
||||
}
|
||||
|
||||
func (r *rtpStatsBase) UpdateFEC(
|
||||
fecPackets uint64,
|
||||
fecBytes uint64,
|
||||
fecPacketsDiscarded uint64,
|
||||
fecPacketsRecovered uint64,
|
||||
) {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
|
||||
if r.endTime != 0 {
|
||||
return
|
||||
}
|
||||
|
||||
r.fecPackets += fecPackets
|
||||
r.fecBytes += fecBytes
|
||||
r.fecPacketsDiscarded += fecPacketsDiscarded
|
||||
r.fecPacketsRecovered += fecPacketsRecovered
|
||||
}
|
||||
|
||||
func (r *rtpStatsBase) UpdateFir(firCount uint32) {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
@@ -503,8 +549,12 @@ func (r *rtpStatsBase) deltaInfo(
|
||||
}
|
||||
if packetsExpected == 0 {
|
||||
deltaInfo = &RTPDeltaInfo{
|
||||
StartTime: time.Unix(0, startTime),
|
||||
EndTime: time.Unix(0, endTime),
|
||||
StartTime: time.Unix(0, startTime),
|
||||
EndTime: time.Unix(0, endTime),
|
||||
FecPackets: uint32(now.fecPackets - then.fecPackets),
|
||||
FecBytes: now.fecBytes - then.fecBytes,
|
||||
FecPacketsDiscarded: uint32(now.fecPacketsDiscarded - then.fecPacketsDiscarded),
|
||||
FecPacketsRecovered: uint32(now.fecPacketsRecovered - then.fecPacketsRecovered),
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -544,6 +594,10 @@ func (r *rtpStatsBase) deltaInfo(
|
||||
PacketsPadding: uint32(packetsPadding),
|
||||
BytesPadding: now.bytesPadding - then.bytesPadding,
|
||||
HeaderBytesPadding: now.headerBytesPadding - then.headerBytesPadding,
|
||||
FecPackets: uint32(now.fecPackets - then.fecPackets),
|
||||
FecBytes: now.fecBytes - then.fecBytes,
|
||||
FecPacketsDiscarded: uint32(now.fecPacketsDiscarded - then.fecPacketsDiscarded),
|
||||
FecPacketsRecovered: uint32(now.fecPacketsRecovered - then.fecPacketsRecovered),
|
||||
PacketsLost: packetsLost,
|
||||
PacketsOutOfOrder: uint32(now.packetsOutOfOrder - then.packetsOutOfOrder),
|
||||
Frames: now.frames - then.frames,
|
||||
@@ -588,6 +642,11 @@ func (r *rtpStatsBase) marshalLogObject(
|
||||
e.AddFloat64("bitratePadding", float64(r.bytesPadding)*8.0/elapsedSeconds)
|
||||
e.AddUint64("headerBytesPadding", r.headerBytesPadding)
|
||||
|
||||
e.AddUint64("fecPackets", r.fecPackets)
|
||||
e.AddUint64("fecBytes", r.fecBytes)
|
||||
e.AddUint64("fecPacketsDiscarded", r.fecPacketsDiscarded)
|
||||
e.AddUint64("fecPacketsRecovered", r.fecPacketsRecovered)
|
||||
|
||||
e.AddUint32("frames", r.frames)
|
||||
e.AddFloat64("frameRate", float64(r.frames)/elapsedSeconds)
|
||||
|
||||
@@ -638,6 +697,11 @@ func (r *rtpStatsBase) toProto(
|
||||
p.BitratePadding = float64(r.bytesPadding) * 8.0 / p.Duration
|
||||
p.HeaderBytesPadding = r.headerBytesPadding
|
||||
|
||||
p.FecPackets = uint32(r.fecPackets)
|
||||
p.FecBytes = r.fecBytes
|
||||
p.FecPacketsDiscarded = uint32(r.fecPacketsDiscarded)
|
||||
p.FecPacketsRecovered = uint32(r.fecPacketsRecovered)
|
||||
|
||||
p.Frames = r.frames
|
||||
p.FrameRate = float64(r.frames) / p.Duration
|
||||
|
||||
@@ -813,6 +877,10 @@ func (r *rtpStatsBase) getSnapshot(startTime int64, extStartSN uint64) snapshot
|
||||
packetsPadding: r.packetsPadding,
|
||||
bytesPadding: r.bytesPadding,
|
||||
headerBytesPadding: r.headerBytesPadding,
|
||||
fecPackets: r.fecPackets,
|
||||
fecBytes: r.fecBytes,
|
||||
fecPacketsDiscarded: r.fecPacketsDiscarded,
|
||||
fecPacketsRecovered: r.fecPacketsRecovered,
|
||||
frames: r.frames,
|
||||
plis: r.plis,
|
||||
firs: r.firs,
|
||||
@@ -853,6 +921,11 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo {
|
||||
bytesPadding := uint64(0)
|
||||
headerBytesPadding := uint64(0)
|
||||
|
||||
fecPackets := uint32(0)
|
||||
fecBytes := uint64(0)
|
||||
fecPacketsDiscarded := uint32(0)
|
||||
fecPacketsRecovered := uint32(0)
|
||||
|
||||
packetsLost := uint32(0)
|
||||
packetsMissing := uint32(0)
|
||||
packetsOutOfOrder := uint32(0)
|
||||
@@ -891,6 +964,11 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo {
|
||||
bytesPadding += deltaInfo.BytesPadding
|
||||
headerBytesPadding += deltaInfo.HeaderBytesPadding
|
||||
|
||||
fecPackets += deltaInfo.FecPackets
|
||||
fecBytes += deltaInfo.FecBytes
|
||||
fecPacketsDiscarded += deltaInfo.FecPacketsDiscarded
|
||||
fecPacketsRecovered += deltaInfo.FecPacketsRecovered
|
||||
|
||||
packetsLost += deltaInfo.PacketsLost
|
||||
packetsMissing += deltaInfo.PacketsMissing
|
||||
packetsOutOfOrder += deltaInfo.PacketsOutOfOrder
|
||||
@@ -925,6 +1003,10 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo {
|
||||
PacketsPadding: packetsPadding,
|
||||
BytesPadding: bytesPadding,
|
||||
HeaderBytesPadding: headerBytesPadding,
|
||||
FecPackets: fecPackets,
|
||||
FecBytes: fecBytes,
|
||||
FecPacketsDiscarded: fecPacketsDiscarded,
|
||||
FecPacketsRecovered: fecPacketsRecovered,
|
||||
PacketsLost: packetsLost,
|
||||
PacketsMissing: packetsMissing,
|
||||
PacketsOutOfOrder: packetsOutOfOrder,
|
||||
|
||||
@@ -223,6 +223,52 @@ func Test_RTPStatsReceiver_Update(t *testing.T) {
|
||||
r.Stop()
|
||||
}
|
||||
|
||||
func Test_RTPStatsReceiver_UpdateFEC(t *testing.T) {
|
||||
r := NewRTPStatsReceiver(RTPStatsParams{})
|
||||
r.SetClockRate(90000)
|
||||
snapshotID := r.NewSnapshotId()
|
||||
|
||||
packet := getPacket(100, 3000, 1000)
|
||||
r.Update(
|
||||
time.Now().UnixNano(),
|
||||
packet.SequenceNumber,
|
||||
packet.Timestamp,
|
||||
packet.Marker,
|
||||
packet.Header.MarshalSize(),
|
||||
len(packet.Payload),
|
||||
0,
|
||||
)
|
||||
r.UpdateFEC(3, 900, 1, 2)
|
||||
|
||||
stats := r.ToProto()
|
||||
require.EqualValues(t, 3, stats.FecPackets)
|
||||
require.EqualValues(t, 900, stats.FecBytes)
|
||||
require.EqualValues(t, 1, stats.FecPacketsDiscarded)
|
||||
require.EqualValues(t, 2, stats.FecPacketsRecovered)
|
||||
|
||||
delta := r.DeltaInfo(snapshotID)
|
||||
require.EqualValues(t, 3, delta.FecPackets)
|
||||
require.EqualValues(t, 900, delta.FecBytes)
|
||||
require.EqualValues(t, 1, delta.FecPacketsDiscarded)
|
||||
require.EqualValues(t, 2, delta.FecPacketsRecovered)
|
||||
|
||||
// FEC-only intervals are reported even if no primary packet arrived.
|
||||
r.UpdateFEC(2, 500, 1, 1)
|
||||
delta = r.DeltaInfo(snapshotID)
|
||||
require.EqualValues(t, 2, delta.FecPackets)
|
||||
require.EqualValues(t, 500, delta.FecBytes)
|
||||
require.EqualValues(t, 1, delta.FecPacketsDiscarded)
|
||||
require.EqualValues(t, 1, delta.FecPacketsRecovered)
|
||||
|
||||
r.Stop()
|
||||
r.UpdateFEC(1, 100, 1, 1)
|
||||
stats = r.ToProto()
|
||||
require.EqualValues(t, 5, stats.FecPackets)
|
||||
require.EqualValues(t, 1400, stats.FecBytes)
|
||||
require.EqualValues(t, 2, stats.FecPacketsDiscarded)
|
||||
require.EqualValues(t, 3, stats.FecPacketsRecovered)
|
||||
}
|
||||
|
||||
func Test_RTPStatsReceiver_Restart(t *testing.T) {
|
||||
clockRate := uint32(90000)
|
||||
r := NewRTPStatsReceiver(RTPStatsParams{})
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
// Copyright 2026 LiveKit, Inc.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package rtptest
|
||||
|
||||
import (
|
||||
"math/rand"
|
||||
|
||||
"github.com/pion/rtp"
|
||||
)
|
||||
|
||||
// GenerateVP8Packets returns deterministic RTP packets with valid VP8 payload descriptors.
|
||||
func GenerateVP8Packets(baseSN uint16, count int, payloadType uint8, ssrc uint32) []rtp.Packet {
|
||||
rng := rand.New(rand.NewSource(int64(baseSN)))
|
||||
packets := make([]rtp.Packet, 0, count)
|
||||
for i := 0; i < count; i++ {
|
||||
payload := make([]byte, 50+rng.Intn(200))
|
||||
rng.Read(payload)
|
||||
payload[0] = 0x10
|
||||
|
||||
sequenceNumber := baseSN + uint16(i)
|
||||
packets = append(packets, rtp.Packet{
|
||||
Header: rtp.Header{
|
||||
Version: 2,
|
||||
PayloadType: payloadType,
|
||||
SequenceNumber: sequenceNumber,
|
||||
Timestamp: 90000 + 3000*uint32(sequenceNumber),
|
||||
SSRC: ssrc,
|
||||
Marker: i == count-1,
|
||||
},
|
||||
Payload: payload,
|
||||
})
|
||||
}
|
||||
return packets
|
||||
}
|
||||
Reference in new issue
Block a user