diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index a2185063f..22cc6fadc 100644 --- a/pkg/sfu/downtrack.go +++ b/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() diff --git a/pkg/sfu/playoutdelay.go b/pkg/sfu/playoutdelay.go new file mode 100644 index 000000000..eaef62853 --- /dev/null +++ b/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 +} diff --git a/pkg/sfu/playoutdelay_test.go b/pkg/sfu/playoutdelay_test.go new file mode 100644 index 000000000..190891034 --- /dev/null +++ b/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) +}