diff --git a/pkg/sfu/buffer/buffer_test.go b/pkg/sfu/buffer/buffer_test.go index e49ae93d5..8b675c855 100644 --- a/pkg/sfu/buffer/buffer_test.go +++ b/pkg/sfu/buffer/buffer_test.go @@ -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}, diff --git a/pkg/sfu/buffer/nack.go b/pkg/sfu/buffer/nack.go index 37cf6a30b..fc45fbb0c 100644 --- a/pkg/sfu/buffer/nack.go +++ b/pkg/sfu/buffer/nack.go @@ -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 +} diff --git a/pkg/sfu/buffer/nack_test.go b/pkg/sfu/buffer/nack_test.go index cb92d071e..5b3f784b5 100644 --- a/pkg/sfu/buffer/nack_test.go +++ b/pkg/sfu/buffer/nack_test.go @@ -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)