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

NACKs with min/max intervals and exponential backoff retries (#425)

Browse Source 
* NACKs with min/max intervals and exponential backoff retries

* simplify
This commit is contained in:
Raja Subramanian
2022-02-09 19:00:03 +05:30
committed by GitHub
parent f4f9df375f
commit aff25afb70
3 changed files with 94 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/buffer/buffer_test.go
+18 -5
View File
@@ -1,6 +1,7 @@
package buffer
import (
"math"
"sync"
"testing"
"time"
@@ -43,8 +44,8 @@ func TestNack(t *testing.T) {
buff.codecType = webrtc.RTPCodecTypeVideo
require.NotNil(t, buff)
var wg sync.WaitGroup
// 3 nacks
wg.Add(3)
// 5 tries
wg.Add(5)
buff.OnFeedback(func(fb []rtcp.Packet) {
for _, pkt := range fb {
switch p := pkt.(type) {
@@ -59,11 +60,17 @@ func TestNack(t *testing.T) {
HeaderExtensions: nil,
Codecs: []webrtc.RTPCodecParameters{vp8Codec},
}, vp8Codec.RTPCodecCapability, Options{})
buff.nacker.SetRTT(0)
rtt := uint32(20)
buff.nacker.SetRTT(rtt)
for i := 0; i < 15; i++ {
if i == 1 {
continue
}
if i < 14 {
time.Sleep(time.Duration(float64(rtt)*math.Pow(backoffFactor, float64(i))+10) * time.Millisecond)
} else {
time.Sleep(500 * time.Millisecond) // even a long wait should not exceed max retries
}
pkt := rtp.Packet{
Header: rtp.Header{SequenceNumber: uint16(i), Timestamp: uint32(i)},
Payload: []byte{0xff, 0xff, 0xff, 0xfd, 0xb4, 0x9f, 0x94, 0x1},
@@ -88,7 +95,7 @@ func TestNack(t *testing.T) {
0: 0,
1: 0,
}
wg.Add(3 * len(expects)) // retry 3 times
wg.Add(5 * len(expects)) // retry 5 times
buff.OnFeedback(func(fb []rtcp.Packet) {
for _, pkt := range fb {
switch p := pkt.(type) {
@@ -112,11 +119,17 @@ func TestNack(t *testing.T) {
HeaderExtensions: nil,
Codecs: []webrtc.RTPCodecParameters{vp8Codec},
}, vp8Codec.RTPCodecCapability, Options{})
buff.nacker.SetRTT(0)
rtt := uint32(30)
buff.nacker.SetRTT(rtt)
for i := 0; i < 15; i++ {
if i > 0 && i < 5 {
continue
}
if i < 14 {
time.Sleep(time.Duration(float64(rtt)*math.Pow(backoffFactor, float64(i))+10) * time.Millisecond)
} else {
time.Sleep(500 * time.Millisecond) // even a long wait should not exceed max retries
}
pkt := rtp.Packet{
Header: rtp.Header{SequenceNumber: uint16(i + 65533), Timestamp: uint32(i)},
Payload: []byte{0xff, 0xff, 0xff, 0xfd, 0xb4, 0x9f, 0x94, 0x1},
pkg/sfu/buffer/nack.go
+74 -23
View File
@@ -1,33 +1,34 @@
package buffer
import (
"math"
"time"
"github.com/pion/rtcp"
)
const maxNackTimes = 3 // Max number of times a packet will be NACKed
const maxNackCache = 100 // Max NACK sn the sfu will keep reference
type nack struct {
seqNum uint16
nacked uint8
lastNackTime time.Time
}
const (
maxTries = 5 // Max number of times a packet will be NACKed
cacheSize = 100 // Max NACK sn the sfu will keep reference
minInterval = 20 * time.Millisecond // minimum interval between NACK tries for the same sequence number
maxInterval = 400 * time.Millisecond // maximum interval between NACK tries for the same sequence number
initialDelay = 10 * time.Millisecond // delay before NACKing a sequence number to allow for out-of-order packets
backoffFactor = float64(1.25)
)
type NackQueue struct {
nacks []*nack
rtt time.Duration
rtt uint32
}
func NewNACKQueue() *NackQueue {
return &NackQueue{
nacks: make([]*nack, 0, maxNackCache),
nacks: make([]*nack, 0, cacheSize),
}
}
func (n *NackQueue) SetRTT(rtt uint32) {
n.rtt = time.Duration(rtt) * time.Millisecond
n.rtt = rtt
}
func (n *NackQueue) Remove(sn uint16) {
@@ -49,7 +50,7 @@ func (n *NackQueue) Push(sn uint16) {
n.nacks = n.nacks[:len(n.nacks)-1]
}
n.nacks = append(n.nacks, &nack{seqNum: sn, nacked: 0, lastNackTime: time.Now()})
n.nacks = append(n.nacks, newNack(sn))
}
func (n *NackQueue) Pairs() ([]rtcp.NackPair, int) {
@@ -69,31 +70,31 @@ func (n *NackQueue) Pairs() ([]rtcp.NackPair, int) {
var np rtcp.NackPair
var nps []rtcp.NackPair
for _, nack := range n.nacks {
if nack.nacked >= maxNackTimes || now.Sub(nack.lastNackTime) < n.rtt {
if nack.nacked >= maxNackTimes {
snsToPurge = append(snsToPurge, nack.seqNum)
}
shouldSend, shouldRemove, sn := nack.getNack(now, n.rtt)
if shouldRemove {
snsToPurge = append(snsToPurge, sn)
continue
}
if !shouldSend {
continue
}
nack.nacked++
nack.lastNackTime = now
numSeqNumsNacked++
if (nack.seqNum - baseSN) > 16 {
if (sn - baseSN) > 16 {
// need a new nack pair
if isPairActive {
nps = append(nps, np)
isPairActive = false
}
baseSN = nack.seqNum
baseSN = sn
np.PacketID = nack.seqNum
np.PacketID = sn
np.LostPackets = 0
isPairActive = true
} else {
np.LostPackets |= 1 << (nack.seqNum - baseSN - 1)
np.LostPackets |= 1 << (sn - baseSN - 1)
}
}
@@ -108,3 +109,53 @@ func (n *NackQueue) Pairs() ([]rtcp.NackPair, int) {
return nps, numSeqNumsNacked
}
// -----------------------------------------------------------------
type nack struct {
seqNum uint16
tries uint8
lastNackedAt time.Time
}
func newNack(sn uint16) *nack {
return &nack{
seqNum: sn,
tries: 0,
lastNackedAt: time.Now(),
}
}
func (n *nack) getNack(now time.Time, rtt uint32) (shouldSend bool, shouldRemove bool, sn uint16) {
sn = n.seqNum
if n.tries >= maxTries {
shouldRemove = true
return
}
var requiredInterval time.Duration
if n.tries > 0 {
// exponentially backoff retries, but cap maximum spacing between retries
requiredInterval := maxInterval
backoffInterval := time.Duration(float64(rtt)*math.Pow(backoffFactor, float64(n.tries-1))) * time.Millisecond
if backoffInterval < requiredInterval {
requiredInterval = backoffInterval
}
}
if requiredInterval < minInterval {
//
// Wait for some time for out-of-order packets before NACKing even if before NACKing first time.
// For subsequent tries, maintain minimum spacing.
//
requiredInterval = minInterval
}
if now.Sub(n.lastNackedAt) < requiredInterval {
return
}
n.tries++
n.lastNackedAt = now
shouldSend = true
return
}
pkg/sfu/buffer/nack_test.go
+2
View File
@@ -2,6 +2,7 @@ package buffer
import (
"testing"
"time"
"github.com/pion/rtcp"
"github.com/stretchr/testify/require"
@@ -73,6 +74,7 @@ func Test_nackQueue_pairs(t *testing.T) {
for _, sn := range tt.args {
n.Push(sn)
}
time.Sleep(100 * time.Millisecond)
got, numSeqNumsNacked := n.Pairs()
require.EqualValues(t, tt.want.pairs, got)
require.Equal(t, tt.want.numSeqNumsNacked, numSeqNumsNacked)