Add dynamic playout delay if PlayoutDelay enabled in the room (#2403)

* Add dynamic playout delay

* type for state

pkg/sfu/downtrack.go

@@ -262,8 +262,7 @@ type DownTrack struct {
 	bytesSent                       atomic.Uint32
 	bytesRetransmitted              atomic.Uint32
 
-	playoutDelayBytes atomic.Value //bytes of marshalled playout delay
-	playoudDelayAcked atomic.Bool
+	playoutDelay *PlayoutDelayController
 
 	pacer pacer.Pacer
 
@@ -318,6 +317,14 @@ func NewDownTrack(params DowntrackParams) (*DownTrack, error) {
 	})
 	d.deltaStatsSenderSnapshotId = d.rtpStats.NewSenderSnapshotId()
 
+	if delay := params.PlayoutDelayLimit; delay.GetEnabled() {
+		var err error
+		d.playoutDelay, err = NewPlayoutDelayController(delay.GetMin(), delay.GetMax(), params.Logger, d.rtpStats)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
 		MimeType:       codecs[0].MimeType, // LK-TODO have to notify on codec change
 		IsFECEnabled:   strings.EqualFold(codecs[0].MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(codecs[0].SDPFmtpLine), "fec"),
@@ -330,23 +337,6 @@ func NewDownTrack(params DowntrackParams) (*DownTrack, error) {
 		}
 	})
 
-	// set initial playout delay to minimum value
-	if d.params.PlayoutDelayLimit.GetEnabled() {
-		maxDelay := uint32(rtpextension.PlayoutDelayDefaultMax)
-		if d.params.PlayoutDelayLimit.GetMax() > 0 {
-			maxDelay = d.params.PlayoutDelayLimit.GetMax()
-		}
-		delay := rtpextension.PlayoutDelayFromValue(
-			uint16(d.params.PlayoutDelayLimit.GetMin()),
-			uint16(maxDelay),
-		)
-		b, err := delay.Marshal()
-		if err == nil {
-			d.playoutDelayBytes.Store(b)
-		} else {
-			d.params.Logger.Errorw("failed to marshal playout delay", err, "playoutDelay", d.params.PlayoutDelayLimit)
-		}
-	}
 	if d.kind == webrtc.RTPCodecTypeVideo {
 		go d.maxLayerNotifierWorker()
 		go d.keyFrameRequester()
@@ -716,9 +706,9 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 	if tp.ddBytes != nil {
 		extensions = []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: tp.ddBytes}}
 	}
-	if d.playoutDelayExtID != 0 && !d.playoudDelayAcked.Load() {
-		if val := d.playoutDelayBytes.Load(); val != nil {
-			extensions = append(extensions, pacer.ExtensionData{ID: uint8(d.playoutDelayExtID), Payload: val.([]byte)})
+	if d.playoutDelayExtID != 0 && d.playoutDelay != nil {
+		if val := d.playoutDelay.GetDelayExtension(hdr.SequenceNumber); val != nil {
+			extensions = append(extensions, pacer.ExtensionData{ID: uint8(d.playoutDelayExtID), Payload: val})
 		}
 	}
 	if d.sequencer != nil {
@@ -1534,7 +1524,11 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 					sal.OnRTCPReceiverReport(d, r)
 				}
 
-				d.playoudDelayAcked.Store(true)
+				if d.playoutDelay != nil {
+					jitterMs := uint64(r.Jitter*1e3) / uint64(d.codec.ClockRate)
+					d.playoutDelay.OnSeqAcked(uint16(r.LastSequenceNumber))
+					d.playoutDelay.SetJitter(uint32(jitterMs))
+				}
 			}
 			if len(rr.Reports) > 0 {
 				d.listenerLock.RLock()

pkg/sfu/playoutdelay.go

@@ -0,0 +1,138 @@
+package sfu
+
+import (
+	"sync"
+	"sync/atomic"
+
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/livekit-server/pkg/sfu/rtpextension"
+	"github.com/livekit/protocol/logger"
+)
+
+type PlayoutDelayState int32
+
+const (
+	PlayoutDelayStateChanged PlayoutDelayState = iota
+	PlayoutDelaySending
+	PlayoutDelayAcked
+
+	jitterLowMultiToDelay  = 10
+	jitterHighMultiToDelay = 15
+	jitterHighThreshold    = 15
+)
+
+func (s PlayoutDelayState) String() string {
+	switch s {
+	case PlayoutDelayStateChanged:
+		return "StateChanged"
+	case PlayoutDelaySending:
+		return "Sending"
+	case PlayoutDelayAcked:
+		return "Acked"
+	}
+	return "Unknown"
+}
+
+type PlayoutDelayController struct {
+	lock               sync.Mutex
+	state              atomic.Int32
+	minDelay, maxDelay uint32
+	currentDelay       uint32
+	extBytes           atomic.Value //[]byte
+	sendingAtSeq       uint16
+	logger             logger.Logger
+	rtpStats           *buffer.RTPStatsSender
+	snapshotID         uint32
+}
+
+func NewPlayoutDelayController(minDelay, maxDelay uint32, logger logger.Logger, rtpStats *buffer.RTPStatsSender) (*PlayoutDelayController, error) {
+	if maxDelay == 0 || maxDelay > rtpextension.PlayoutDelayDefaultMax {
+		maxDelay = rtpextension.PlayoutDelayDefaultMax
+	}
+	c := &PlayoutDelayController{
+		currentDelay: minDelay,
+		minDelay:     minDelay,
+		maxDelay:     maxDelay,
+		logger:       logger,
+		rtpStats:     rtpStats,
+		snapshotID:   rtpStats.NewSenderSnapshotId(),
+	}
+	return c, c.createExtData()
+}
+
+func (c *PlayoutDelayController) SetJitter(jitter uint32) {
+	deltaInfo := c.rtpStats.DeltaInfoSender(c.snapshotID)
+	var nackPercent uint32
+	if deltaInfo != nil && deltaInfo.Packets > 0 {
+		nackPercent = deltaInfo.Nacks * 100 / deltaInfo.Packets
+	}
+
+	c.lock.Lock()
+	multi := jitterLowMultiToDelay
+	if jitter >= jitterHighThreshold {
+		multi = jitterHighMultiToDelay
+	}
+	targetDelay := jitter * uint32(multi)
+	if nackPercent > 60 {
+		targetDelay += (nackPercent - 60) * 2
+	}
+
+	// increase delay quickly, decrease slowly to make fps more stable
+	if targetDelay > c.currentDelay {
+		targetDelay = (targetDelay-c.currentDelay)*3/4 + c.currentDelay
+	} else {
+		targetDelay = c.currentDelay - (c.currentDelay-targetDelay)/5
+	}
+	if targetDelay < c.minDelay {
+		targetDelay = c.minDelay
+	}
+	if targetDelay > c.maxDelay {
+		targetDelay = c.maxDelay
+	}
+	if c.currentDelay == targetDelay {
+		c.lock.Unlock()
+		return
+	}
+	c.currentDelay = targetDelay
+	c.lock.Unlock()
+	c.createExtData()
+}
+
+func (c *PlayoutDelayController) OnSeqAcked(seq uint16) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if PlayoutDelayState(c.state.Load()) == PlayoutDelaySending && (seq-c.sendingAtSeq) < 0x8000 {
+		c.state.Store(int32(PlayoutDelayAcked))
+	}
+}
+
+func (c *PlayoutDelayController) GetDelayExtension(seq uint16) []byte {
+	switch PlayoutDelayState(c.state.Load()) {
+	case PlayoutDelayStateChanged:
+		c.lock.Lock()
+		c.state.Store(int32(PlayoutDelaySending))
+		c.sendingAtSeq = seq
+		c.lock.Unlock()
+		return c.extBytes.Load().([]byte)
+	case PlayoutDelaySending:
+		return c.extBytes.Load().([]byte)
+	case PlayoutDelayAcked:
+		return nil
+	}
+	return nil
+}
+
+func (c *PlayoutDelayController) createExtData() error {
+	delay := rtpextension.PlayoutDelayFromValue(
+		uint16(c.currentDelay),
+		uint16(c.maxDelay),
+	)
+	b, err := delay.Marshal()
+	if err == nil {
+		c.extBytes.Store(b)
+		c.state.Store(int32(PlayoutDelayStateChanged))
+	} else {
+		c.logger.Errorw("failed to marshal playout delay", err, "playoutDelay", delay)
+	}
+	return err
+}

pkg/sfu/playoutdelay_test.go

@@ -0,0 +1,61 @@
+package sfu
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/livekit-server/pkg/sfu/rtpextension"
+	"github.com/livekit/protocol/logger"
+)
+
+func TestPlayoutDelay(t *testing.T) {
+	stats := buffer.NewRTPStatsSender(buffer.RTPStatsParams{ClockRate: 900000, Logger: logger.GetLogger()})
+	c, err := NewPlayoutDelayController(100, 1000, logger.GetLogger(), stats)
+	require.NoError(t, err)
+
+	ext := c.GetDelayExtension(100)
+	playoutDelayEqual(t, ext, 100, 1000)
+
+	ext = c.GetDelayExtension(105)
+	playoutDelayEqual(t, ext, 100, 1000)
+
+	// seq acked before delay changed
+	c.OnSeqAcked(65534)
+	ext = c.GetDelayExtension(105)
+	playoutDelayEqual(t, ext, 100, 1000)
+
+	c.OnSeqAcked(90)
+	ext = c.GetDelayExtension(105)
+	playoutDelayEqual(t, ext, 100, 1000)
+
+	// seq acked, no extension sent for new packet
+	c.OnSeqAcked(103)
+	ext = c.GetDelayExtension(106)
+	require.Nil(t, ext)
+
+	// delay on change(can't go below min), no extension sent
+	c.SetJitter(0)
+	ext = c.GetDelayExtension(107)
+	require.Nil(t, ext)
+
+	// delay changed, generate new extension to send
+	c.SetJitter(50)
+	ext = c.GetDelayExtension(108)
+	var delay rtpextension.PlayOutDelay
+	require.NoError(t, delay.Unmarshal(ext))
+	require.Greater(t, delay.Min, uint16(100))
+
+	// can't go above max
+	c.SetJitter(10000)
+	ext = c.GetDelayExtension(109)
+	playoutDelayEqual(t, ext, 1000, 1000)
+}
+
+func playoutDelayEqual(t *testing.T, data []byte, min, max uint16) {
+	var delay rtpextension.PlayOutDelay
+	require.NoError(t, delay.Unmarshal(data))
+	require.Equal(t, min, delay.Min)
+	require.Equal(t, max, delay.Max)
+}