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

Process RTX packets for TWCC (#406)

Browse Source 
The SRTP replay detection was disabled recently.
But, they were effectively getting dropped in `sfu.bucket`.

Doing two things with RTX packets in this PR
1. Update stats - add to packet count and bytes
2. Process header extension - to process TWCC
This commit is contained in:
Raja Subramanian
2022-02-05 23:25:53 +05:30
committed by GitHub
parent 7f2a5ae036
commit 4a9b844781
2 changed files with 37 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/buffer/bucket.go
+2 -2
View File
@@ -33,8 +33,8 @@ func (b *Bucket) AddPacket(pkt []byte) ([]byte, error) {
b.init = true
}
diff := sn - b.headSN
if diff > (1 << 15) {
// out-of-order
if diff == 0 || diff > (1<<15) {
// duplicate of lsat packet or out-of-order
return b.set(sn, pkt)
}
b.headSN = sn
pkg/sfu/buffer/buffer.go
+35 -13
View File
@@ -272,24 +272,44 @@ func (b *Buffer) OnClose(fn func()) {
}
func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
isRTX := false
pb, err := b.bucket.AddPacket(pkt)
if err != nil {
if err != ErrRTXPacket {
b.logger.Warnw("could not add RTP packet to bucket", err)
return
} else {
isRTX = true
}
return
}
var p rtp.Packet
if err := p.Unmarshal(pb); err != nil {
if isRTX {
err = p.Unmarshal(pkt)
} else {
err = p.Unmarshal(pb)
}
if err != nil {
b.logger.Warnw("error unmarshaling RTP packet", err)
return
}
b.updateStreamState(&p, len(pkt), arrivalTime)
b.updateStreamState(&p, len(pkt), arrivalTime, isRTX)
b.processHeaderExtensions(&p, arrivalTime)
if isRTX {
//
// Run RTX packets through
// 1. state update - to update stats
// 2. TWCC just in case remote side is retransmitting an old packet for probing
//
// But, do not forward those packets
//
return
}
ep, temporalLayer := b.getExtPacket(pb, &p, arrivalTime)
if ep == nil {
return
@@ -305,7 +325,7 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
b.doReports(arrivalTime)
}
func (b *Buffer) updateStreamState(p *rtp.Packet, pktSize int, arrivalTime int64) {
func (b *Buffer) updateStreamState(p *rtp.Packet, pktSize int, arrivalTime int64, isRTX bool) {
sn := p.SequenceNumber
if b.stats.PacketCount == 0 {
@@ -344,17 +364,19 @@ func (b *Buffer) updateStreamState(p *rtp.Packet, pktSize int, arrivalTime int64
b.stats.PacketCount++
b.stats.TotalBytes += uint64(pktSize)
// jitter
arrival := uint32(arrivalTime / 1e6 * int64(b.clockRate/1e3))
transit := arrival - p.Timestamp
if b.lastTransit != 0 {
d := int32(transit - b.lastTransit)
if d < 0 {
d = -d
if !isRTX {
// jitter
arrival := uint32(arrivalTime / 1e6 * int64(b.clockRate/1e3))
transit := arrival - p.Timestamp
if b.lastTransit != 0 {
d := int32(transit - b.lastTransit)
if d < 0 {
d = -d
}
b.stats.Jitter += (float64(d) - b.stats.Jitter) / 16
}
b.stats.Jitter += (float64(d) - b.stats.Jitter) / 16
b.lastTransit = transit
}
b.lastTransit = transit
}
func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) {