diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index eefa1c0e3..431f6aaa1 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -99,6 +99,8 @@ type MediaTrackParams struct { EnableRTPStreamRestartDetection bool UpdateTrackInfoByVideoSizeChange bool ForceBackupCodecPolicySimulcast bool + EnableVideoCaching bool + VideoCachingMaxDuration time.Duration } func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack { @@ -407,6 +409,18 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe return newCodec, false } + receiverOpts := []sfu.ReceiverOpts{ + sfu.WithPliThrottleConfig(t.params.PLIThrottleConfig), + sfu.WithAudioConfig(t.params.AudioConfig), + sfu.WithLoadBalanceThreshold(20), + sfu.WithForwardStats(t.params.ForwardStats), + sfu.WithEnableRTPStreamRestartDetection(t.params.EnableRTPStreamRestartDetection), + } + if t.params.EnableVideoCaching { + // VideoCachingMaxDuration <= 0 falls back to sfu.DefaultVideoFrameCacheMaxDuration in WithVideoFrameCache + receiverOpts = append(receiverOpts, sfu.WithVideoFrameCache(t.params.VideoCachingMaxDuration)) + } + newWR := sfu.NewWebRTCReceiver( receiver, track, @@ -414,11 +428,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe LoggerWithCodecMime(t.params.Logger, mimeType), t.params.OnRTCP, t.params.VideoConfig.StreamTrackerManager, - sfu.WithPliThrottleConfig(t.params.PLIThrottleConfig), - sfu.WithAudioConfig(t.params.AudioConfig), - sfu.WithLoadBalanceThreshold(20), - sfu.WithForwardStats(t.params.ForwardStats), - sfu.WithEnableRTPStreamRestartDetection(t.params.EnableRTPStreamRestartDetection), + receiverOpts..., ) newWR.OnCloseHandler(func() { t.MediaTrackReceiver.SetClosing(false) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index ebe59cc58..db8de5823 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -226,6 +226,8 @@ type ParticipantParams struct { ForceBackupCodecPolicySimulcast bool DisableTransceiverReuseForE2EE bool EnableStartAtDesiredQuality bool + EnableVideoCaching bool + VideoCachingMaxDuration time.Duration } type ParticipantImpl struct { @@ -3382,6 +3384,8 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, ti *livekit.TrackInfo) EnableRTPStreamRestartDetection: p.params.EnableRTPStreamRestartDetection, UpdateTrackInfoByVideoSizeChange: p.params.UseOneShotSignallingMode, ForceBackupCodecPolicySimulcast: p.params.ForceBackupCodecPolicySimulcast, + EnableVideoCaching: p.params.EnableVideoCaching, + VideoCachingMaxDuration: p.params.VideoCachingMaxDuration, }, ti) mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 5b376a490..d93493107 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -22,6 +22,7 @@ import ( "github.com/pion/rtcp" "github.com/pion/rtp" "github.com/pion/webrtc/v4" + "go.uber.org/atomic" sutils "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/mediatransportutil/pkg/bucket" @@ -72,6 +73,9 @@ type Buffer struct { primaryBufferForRTX *Buffer rtxPktBuf []byte + + // number of PLIs actually forwarded to the publisher (past the throttle gate, see sendPLI) + pliForwardedCount atomic.Uint32 } func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer { @@ -335,7 +339,6 @@ func (b *Buffer) sendPLI() { return } - b.logger.Debugw("send pli", "mediaSSRC", ssrc) pli := []rtcp.Packet{ &rtcp.PictureLossIndication{ SenderSSRC: ssrc, @@ -344,10 +347,17 @@ func (b *Buffer) sendPLI() { } if cb := b.getOnRtcpFeedback(); cb != nil { + pliForwardedCount := b.pliForwardedCount.Inc() cb(pli) + b.logger.Debugw("send pli", "mediaSSRC", ssrc, "pliForwardedCount", pliForwardedCount) } } +// PLIForwardedCount returns the number of PLIs actually forwarded to the publisher (past the throttle gate). +func (b *Buffer) PLIForwardedCount() uint32 { + return b.pliForwardedCount.Load() +} + func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) []rtcp.Packet { b.BufferBase.HandleIncomingPacketLocked( rawPkt, diff --git a/pkg/sfu/buffer/buffer_base.go b/pkg/sfu/buffer/buffer_base.go index 68980ea05..c79a9b0f0 100644 --- a/pkg/sfu/buffer/buffer_base.go +++ b/pkg/sfu/buffer/buffer_base.go @@ -89,10 +89,16 @@ type BufferProvider interface { SetPaused(paused bool) SendPLI(force bool) + // PLIForwardedCount returns the number of PLIs actually forwarded to the publisher (past the throttle gate). + PLIForwardedCount() uint32 ReadExtended(buf []byte) (*ExtPacket, error) GetPacket(buf []byte, esn uint64) (int, error) + EnableVideoFrameCache(maxDuration time.Duration) + GetVideoFrameCache() ([]*ExtPacket, bool) + GetPacketsAfter(afterSN uint64) ([]*ExtPacket, bool) + GetAudioLevel() (float64, bool) GetTemporalLayerFpsForSpatial(layer int32) []float32 GetStats() *livekit.RTPStats @@ -200,6 +206,16 @@ type BufferBase struct { ddExtID uint8 ddParser *DependencyDescriptorParser + // video frame cache: mark the most recent key frame so the current group-of-pictures can be read + // back from the retransmit bucket (see EnableVideoFrameCache / GetVideoFrameCache). No packets are copied here - only + // the key-frame boundary is tracked. + videoFrameCacheEnabled bool + videoFrameCacheMaxDuration time.Duration // optional bound on key-frame interval (0 = bucket-bounded only) + videoFrameCacheHasKeyFrame bool + videoFrameCacheKeyFrameSN uint64 // ext sequence number of the current GOP's first key-frame packet + videoFrameCacheKeyFrameTS uint64 // ext timestamp of the current GOP's key frame + videoFrameCacheLatestTS uint64 // maximum ext timestamp seen in the current GOP (resets on a new key frame) + isPaused bool frameRateCalculator [DefaultMaxLayerSpatial + 1]FrameRateCalculator frameRateCalculated bool @@ -541,6 +557,8 @@ func (b *BufferBase) restartStreamLocked(reason string, isDetected bool) { b.StopKeyFrameSeeder() b.stopRTPStats("stream-restart") b.flushExtPacketsLocked() + // the marked GOP references the pre-restart sequence-number base / evicted bucket contents + b.videoFrameCacheHasKeyFrame = false // restart b.snRangeMap = utils.NewRangeMap[uint64, uint64](100) @@ -643,6 +661,170 @@ func (b *BufferBase) ReadExtended(buf []byte) (*ExtPacket, error) { } } +// EnableVideoFrameCache turns on video frame cache tracking for this (video) buffer: the most recent key frame is marked +// so the current group-of-pictures can be read back from the retransmit bucket via GetVideoFrameCache. No +// packets are copied - only the key-frame boundary is tracked. maxDuration bounds the served +// key-frame interval AND drives the retransmit bucket to retain that much history (see +// maybeGrowBucket), so the key frame is not evicted before it can be read; <= 0 keeps the default +// ~1s retransmit window (and bounds the GOP to it). No-op for audio buffers. +func (b *BufferBase) EnableVideoFrameCache(maxDuration time.Duration) { + b.Lock() + defer b.Unlock() + + b.videoFrameCacheEnabled = true + b.videoFrameCacheMaxDuration = maxDuration + b.logger.Debugw("video frame cache enabled on buffer", "maxDuration", maxDuration) +} + +// markVideoFrameCacheLocked records the key-frame boundary of the current GOP and tracks its span. Caller holds +// the lock. +func (b *BufferBase) markVideoFrameCacheLocked(ep *ExtPacket) { + if ep == nil || ep.Packet == nil || len(ep.Packet.Payload) == 0 { + return + } + if ep.IsKeyFrame && (!b.videoFrameCacheHasKeyFrame || ep.ExtTimestamp != b.videoFrameCacheKeyFrameTS) { + // a new key frame starts a new GOP; remember its first packet's sequence number and reset the + // span to the key frame so a stale packet from the previous GOP cannot stretch it + b.videoFrameCacheKeyFrameSN = ep.ExtSequenceNumber + b.videoFrameCacheKeyFrameTS = ep.ExtTimestamp + b.videoFrameCacheLatestTS = ep.ExtTimestamp + b.videoFrameCacheHasKeyFrame = true + b.logger.Debugw("video frame cache: marked key frame", "keyFrameSN", b.videoFrameCacheKeyFrameSN, "keyFrameTS", b.videoFrameCacheKeyFrameTS) + return + } + // track the maximum timestamp seen in the current GOP (not the last-written one) so an + // out-of-order, older packet arriving last cannot shrink the measured span and let GetVideoFrameCache serve + // more than videoFrameCacheMaxDuration. The head packet's timestamp is always <= videoFrameCacheLatestTS, so the + // duration gate in GetVideoFrameCache strictly bounds the served GOP. + if ep.ExtTimestamp > b.videoFrameCacheLatestTS { + b.videoFrameCacheLatestTS = ep.ExtTimestamp + } +} + +// GetVideoFrameCache reads the packets of the current group-of-pictures (from the most recent key frame up to +// the latest packet) back from the retransmit bucket, so a newly attached relay / down track can be +// bootstrapped without requesting a fresh key frame (PLI). Returns (nil, false) when the cache is +// disabled, no key frame has been marked, the key-frame interval exceeds the configured bound, or +// the key frame is no longer retained in the bucket (the GOP length is ultimately bounded by the +// bucket capacity). Lost packets within the GOP are skipped. +// +// The packets are returned as ExtPackets reconstructed from the bucket bytes so they can be replayed +// through the normal forward path (WriteRTP): ExtSequenceNumber comes from the bucket key and +// ExtTimestamp / IsKeyFrame are derived from the marked key frame. The dependency descriptor is not +// reconstructed (SVC replay is not supported here). The returned packets are self-contained copies. +func (b *BufferBase) GetVideoFrameCache() ([]*ExtPacket, bool) { + b.Lock() + defer b.Unlock() + + if !b.videoFrameCacheEnabled || !b.videoFrameCacheHasKeyFrame || b.bucket == nil { + b.logger.Debugw( + "video frame cache miss: not ready", + "videoFrameCacheEnabled", b.videoFrameCacheEnabled, + "videoFrameCacheHasKeyFrame", b.videoFrameCacheHasKeyFrame, + "hasBucket", b.bucket != nil, + ) + return nil, false + } + + if b.videoFrameCacheMaxDuration > 0 && b.clockRate > 0 { + maxTicks := uint64(b.videoFrameCacheMaxDuration.Seconds() * float64(b.clockRate)) + if b.videoFrameCacheLatestTS > b.videoFrameCacheKeyFrameTS+maxTicks { + // key-frame interval longer than the bound - too old to serve a complete replay + b.logger.Debugw( + "video frame cache miss: key-frame interval exceeds bound", + "keyFrameTS", b.videoFrameCacheKeyFrameTS, + "latestTS", b.videoFrameCacheLatestTS, + "spanTicks", b.videoFrameCacheLatestTS-b.videoFrameCacheKeyFrameTS, + "maxTicks", maxTicks, + ) + return nil, false + } + } + + headSN := uint64(b.bucket.HeadSequenceNumber()) + if headSN < b.videoFrameCacheKeyFrameSN { + b.logger.Debugw("video frame cache miss: head behind key frame", "headSN", headSN, "keyFrameSN", b.videoFrameCacheKeyFrameSN) + return nil, false + } + + pkts := b.reconstructPacketsLocked(b.videoFrameCacheKeyFrameSN, headSN) + // the key frame itself must be present (its first packet), otherwise the GOP cannot be served + if len(pkts) == 0 || pkts[0].ExtSequenceNumber != b.videoFrameCacheKeyFrameSN { + var firstSN uint64 + if len(pkts) > 0 { + firstSN = pkts[0].ExtSequenceNumber + } + b.logger.Debugw( + "video frame cache miss: key frame evicted from bucket", + "keyFrameSN", b.videoFrameCacheKeyFrameSN, + "headSN", headSN, + "bucketCapacity", b.bucket.Capacity(), + "reconstructed", len(pkts), + "firstSN", firstSN, + ) + return nil, false + } + return pkts, true +} + +// GetPacketsAfter reads the packets newer than afterSN (exclusive) up to the current head from the +// retransmit bucket, reconstructed as ExtPackets like GetVideoFrameCache. It is used to catch a video frame cache replay up to +// the live forwarding point. Returns (nil, false) when the cache is disabled or nothing newer is +// retained. +func (b *BufferBase) GetPacketsAfter(afterSN uint64) ([]*ExtPacket, bool) { + b.Lock() + defer b.Unlock() + + if !b.videoFrameCacheEnabled || !b.videoFrameCacheHasKeyFrame || b.bucket == nil { + return nil, false + } + headSN := uint64(b.bucket.HeadSequenceNumber()) + if headSN <= afterSN { + return nil, false + } + pkts := b.reconstructPacketsLocked(afterSN+1, headSN) + if len(pkts) == 0 { + return nil, false + } + return pkts, true +} + +// reconstructPacketsLocked builds self-contained ExtPackets for the sequence-number range +// [fromSN, headSN] from the retransmit bucket. Lost packets are skipped. ExtTimestamp is +// reconstructed relative to the marked key frame (a GOP spans well under one 32-bit timestamp wrap) +// and IsKeyFrame flags packets at the key-frame timestamp. The dependency descriptor is not +// reconstructed. Caller holds the lock. +func (b *BufferBase) reconstructPacketsLocked(fromSN, headSN uint64) []*ExtPacket { + keyFrameRTPTS := uint32(b.videoFrameCacheKeyFrameTS) + var pkts []*ExtPacket + buf := make([]byte, bucket.RTPMaxPktSize) + for sn := fromSN; sn <= headSN; sn++ { + n, err := b.bucket.GetPacket(buf, sn) + if err != nil { + continue // lost packet, skip + } + // copy out of the reused read buffer so the parsed packet is self-contained + raw := make([]byte, n) + copy(raw, buf[:n]) + p := &rtp.Packet{} + if err := p.Unmarshal(raw); err != nil { + continue + } + + extTS := b.videoFrameCacheKeyFrameTS + uint64(p.Timestamp-keyFrameRTPTS) + pkts = append(pkts, &ExtPacket{ + VideoLayer: VideoLayer{Spatial: InvalidLayerSpatial, Temporal: InvalidLayerTemporal}, + Arrival: mono.UnixNano(), + ExtSequenceNumber: sn, + ExtTimestamp: extTS, + Packet: p, + IsKeyFrame: extTS == b.videoFrameCacheKeyFrameTS, + RawPacket: raw, + }) + } + return pkts +} + func (b *BufferBase) SetPLIThrottle(duration int64) { b.Lock() defer b.Unlock() @@ -909,6 +1091,9 @@ func (b *BufferBase) HandleIncomingPacketLocked( return 0, errors.New("could not get ext packet") } b.extPackets.PushBack(ep) + if b.videoFrameCacheEnabled && b.codecType == webrtc.RTPCodecTypeVideo { + b.markVideoFrameCacheLocked(ep) + } b.readCond.Broadcast() if b.extPackets.Len() > b.bucket.Capacity() { @@ -1224,6 +1409,25 @@ func (b *BufferBase) flushExtPacketsLocked() { b.extPackets.Clear() } +// bucketGrowTarget computes how many packets the retransmit bucket should retain (targetPkts) and +// the cap that allows growing to it (effectiveMaxPkts), given the measured packets-per-second. +// +// Normally the target is ~1s of packets (the NACK / retransmit window), bounded by maxPkts. When +// videoFrameCacheSizing is set (the video frame cache is enabled with a positive duration), the bucket must retain +// the whole GOP duration plus ~0.5s margin so the key frame is not evicted before it is at most +// videoFrameCacheMaxDuration old; the cap is raised to fit since the default maxPkts is only ~1s worth. +func bucketGrowTarget(pps, maxPkts int, videoFrameCacheSizing bool, videoFrameCacheMaxDuration time.Duration) (targetPkts, effectiveMaxPkts int) { + targetPkts = pps + effectiveMaxPkts = maxPkts + if videoFrameCacheSizing && videoFrameCacheMaxDuration > 0 { + targetPkts = int(float64(pps)*videoFrameCacheMaxDuration.Seconds()) + pps/2 + if targetPkts > effectiveMaxPkts { + effectiveMaxPkts = targetPkts + } + } + return +} + func (b *BufferBase) maybeGrowBucket(now int64) { if now-b.lastBucketCapCheckAt < bucketCapCheckInterval { return @@ -1241,31 +1445,45 @@ func (b *BufferBase) maybeGrowBucket(now int64) { if b.codecType == webrtc.RTPCodecTypeAudio { maxPkts = b.params.MaxAudioPkts } + + // when the video frame cache is enabled the bucket must retain the whole configured GOP + // duration (the key frame must survive until GetVideoFrameCache reads it), which is more than the normal + // ~1s retransmit window. In that case the target / cap are computed below from pps; otherwise + // keep the original fast path. + videoFrameCacheSizing := b.codecType == webrtc.RTPCodecTypeVideo && b.videoFrameCacheEnabled && b.videoFrameCacheMaxDuration > 0 + if cap >= maxPkts && !videoFrameCacheSizing { + return + } + + deltaInfo := b.rtpStats.DeltaInfo(b.ppsSnapshotId) + if deltaInfo == nil { + return + } + duration := deltaInfo.EndTime.Sub(deltaInfo.StartTime) + if duration < 500*time.Millisecond { + return + } + pps := int(time.Duration(deltaInfo.Packets) * time.Second / duration) + + targetPkts, maxPkts := bucketGrowTarget(pps, maxPkts, videoFrameCacheSizing, b.videoFrameCacheMaxDuration) if cap >= maxPkts { return } oldCap := cap - if deltaInfo := b.rtpStats.DeltaInfo(b.ppsSnapshotId); deltaInfo != nil { - duration := deltaInfo.EndTime.Sub(deltaInfo.StartTime) - if duration < 500*time.Millisecond { - return - } - - pps := int(time.Duration(deltaInfo.Packets) * time.Second / duration) - for pps > cap && cap < maxPkts { - cap = b.bucket.Grow() - } - if cap > oldCap { - b.logger.Infow( - "grow bucket", - "from", oldCap, - "to", cap, - "pps", pps, - "deltaInfo", deltaInfo, - "rtpStats", b.rtpStats, - ) - } + for targetPkts > cap && cap < maxPkts { + cap = b.bucket.Grow() + } + if cap > oldCap { + b.logger.Infow( + "grow bucket", + "from", oldCap, + "to", cap, + "pps", pps, + "targetPkts", targetPkts, + "deltaInfo", deltaInfo, + "rtpStats", b.rtpStats, + ) } }() } diff --git a/pkg/sfu/buffer/videoframecache_test.go b/pkg/sfu/buffer/videoframecache_test.go new file mode 100644 index 000000000..f247604b0 --- /dev/null +++ b/pkg/sfu/buffer/videoframecache_test.go @@ -0,0 +1,233 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package buffer + +import ( + "testing" + "time" + + "github.com/pion/rtp" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + + "github.com/livekit/mediatransportutil/pkg/bucket" + "github.com/livekit/protocol/logger" +) + +func newVideoFrameCacheTestBuffer(maxDuration time.Duration) *BufferBase { + b := &BufferBase{ + codecType: webrtc.RTPCodecTypeVideo, + clockRate: 90000, + logger: logger.GetLogger(), + } + b.bucket = bucket.NewBucket[uint64, uint16](256, bucket.RTPMaxPktSize, bucket.RTPSeqNumOffset) + b.EnableVideoFrameCache(maxDuration) + return b +} + +// addBucketPacket marshals an RTP packet and stores it in the bucket keyed by ext sequence number. +func addBucketPacket(t *testing.T, b *BufferBase, extSN uint64, ts uint32) { + p := rtp.Packet{ + Header: rtp.Header{Version: 2, PayloadType: 96, SequenceNumber: uint16(extSN), Timestamp: ts, SSRC: 123}, + Payload: []byte{1, 2, 3, 4}, + } + raw, err := p.Marshal() + require.NoError(t, err) + _, err = b.bucket.AddPacketWithSequenceNumber(raw, extSN) + require.NoError(t, err) +} + +func videoFrameCacheMarkPkt(extSN, extTS uint64, keyFrame bool) *ExtPacket { + return &ExtPacket{ + ExtSequenceNumber: extSN, + ExtTimestamp: extTS, + IsKeyFrame: keyFrame, + Packet: &rtp.Packet{Payload: []byte{1, 2, 3, 4}}, + } +} + +func TestVideoFrameCacheReadsFromBucket(t *testing.T) { + b := newVideoFrameCacheTestBuffer(0) + + // store SN 100..104 in the bucket + for i := uint64(0); i < 5; i++ { + addBucketPacket(t, b, 100+i, uint32(1000*(i+1))) + } + + // no key frame marked yet -> not available + _, ok := b.GetVideoFrameCache() + require.False(t, ok) + + // mark: delta at 100, key frame at 101, deltas after + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, false)) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(101, 2000, true)) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(102, 3000, false)) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(103, 4000, false)) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(104, 5000, false)) + + // GOP is [key frame .. head] = 101..104 + pkts, ok := b.GetVideoFrameCache() + require.True(t, ok) + require.Len(t, pkts, 4) + require.Equal(t, uint16(101), pkts[0].Packet.SequenceNumber) + require.Equal(t, uint64(101), pkts[0].ExtSequenceNumber) + require.True(t, pkts[0].IsKeyFrame) // first packet is at the key-frame timestamp + require.False(t, pkts[1].IsKeyFrame) + require.Equal(t, uint16(104), pkts[3].Packet.SequenceNumber) + + // a new key frame moves the boundary + addBucketPacket(t, b, 105, 6000) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(105, 6000, true)) + pkts, ok = b.GetVideoFrameCache() + require.True(t, ok) + require.Len(t, pkts, 1) + require.Equal(t, uint16(105), pkts[0].Packet.SequenceNumber) + require.True(t, pkts[0].IsKeyFrame) +} + +func TestVideoFrameCacheGetPacketsAfter(t *testing.T) { + b := newVideoFrameCacheTestBuffer(0) + for i := uint64(0); i < 5; i++ { + addBucketPacket(t, b, 100+i, uint32(1000*(i+1))) + } + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, true)) + for i := uint64(1); i < 5; i++ { + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100+i, 1000*(i+1), false)) + } + + // everything after 102 is 103, 104 + more, ok := b.GetPacketsAfter(102) + require.True(t, ok) + require.Len(t, more, 2) + require.Equal(t, uint64(103), more[0].ExtSequenceNumber) + require.Equal(t, uint64(104), more[1].ExtSequenceNumber) + + // caught up: nothing after the head + _, ok = b.GetPacketsAfter(104) + require.False(t, ok) +} + +func TestVideoFrameCacheSkipsLostPackets(t *testing.T) { + b := newVideoFrameCacheTestBuffer(0) + + // store the key frame and a later packet, leaving a gap at 101 + addBucketPacket(t, b, 100, 1000) + addBucketPacket(t, b, 102, 3000) + + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, true)) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(102, 3000, false)) + + pkts, ok := b.GetVideoFrameCache() + require.True(t, ok) + require.Len(t, pkts, 2) // the missing 101 is skipped + require.Equal(t, uint16(100), pkts[0].Packet.SequenceNumber) + require.Equal(t, uint16(102), pkts[1].Packet.SequenceNumber) +} + +func TestVideoFrameCacheKeyFrameEvicted(t *testing.T) { + b := newVideoFrameCacheTestBuffer(0) + + // mark a key frame at a sequence number that is not in the bucket + addBucketPacket(t, b, 200, 1000) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 500, true)) // 100 was never stored / evicted + b.videoFrameCacheLatestTS = 1000 + + _, ok := b.GetVideoFrameCache() + require.False(t, ok) +} + +func TestVideoFrameCacheDurationBound(t *testing.T) { + // 2s cap at 90kHz -> 180000 ticks + b := newVideoFrameCacheTestBuffer(2 * time.Second) + addBucketPacket(t, b, 100, 1000) + + const kfTS = uint64(1_000_000) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, kfTS, true)) + + // within the bound + b.videoFrameCacheLatestTS = kfTS + 90000 // +1s + _, ok := b.GetVideoFrameCache() + require.True(t, ok) + + // beyond the bound -> not served + b.videoFrameCacheLatestTS = kfTS + 180001 // > 2s + _, ok = b.GetVideoFrameCache() + require.False(t, ok) +} + +func TestVideoFrameCacheSpanUsesMaxTimestamp(t *testing.T) { + b := newVideoFrameCacheTestBuffer(0) + + // key frame at 1000, then a later packet at 1200 advances the span + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, true)) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(101, 1200, false)) + require.Equal(t, uint64(1200), b.videoFrameCacheLatestTS) + + // an out-of-order, older packet arriving last must not shrink the measured span + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(102, 1100, false)) + require.Equal(t, uint64(1200), b.videoFrameCacheLatestTS) + + // a new key frame resets the span to itself (a stale packet cannot stretch the new GOP) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(103, 5000, true)) + require.Equal(t, uint64(5000), b.videoFrameCacheLatestTS) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(104, 4000, false)) // stale, older than the new key frame + require.Equal(t, uint64(5000), b.videoFrameCacheLatestTS) +} + +func TestBucketGrowTarget(t *testing.T) { + const pps = 600 + const maxPkts = 500 // default ~1s cap + + // video frame cache sizing off -> unchanged original behavior: target is ~1s (pps), cap stays maxPkts + target, cap := bucketGrowTarget(pps, maxPkts, false, 0) + require.Equal(t, pps, target) + require.Equal(t, maxPkts, cap) + + // videoFrameCacheMaxDuration <= 0 is treated as not sizing even if the flag is set + target, cap = bucketGrowTarget(pps, maxPkts, true, 0) + require.Equal(t, pps, target) + require.Equal(t, maxPkts, cap) + + // video frame cache sizing on, 2s: target is pps*2 + 0.5s margin, and the cap is raised to fit + target, cap = bucketGrowTarget(pps, maxPkts, true, 2*time.Second) + require.Equal(t, pps*2+pps/2, target) // 1500 + require.Equal(t, target, cap) // raised above the 500 default + require.Greater(t, cap, maxPkts) + + // video frame cache sizing on but the target fits within maxPkts -> cap is not lowered + target, cap = bucketGrowTarget(100, 1000, true, 2*time.Second) + require.Equal(t, 100*2+100/2, target) // 250 + require.Equal(t, 1000, cap) // unchanged, target < maxPkts + + // scales with duration + target1s, _ := bucketGrowTarget(pps, maxPkts, true, time.Second) + target3s, _ := bucketGrowTarget(pps, maxPkts, true, 3*time.Second) + require.Greater(t, target3s, target1s) +} + +func TestVideoFrameCacheDisabled(t *testing.T) { + b := &BufferBase{codecType: webrtc.RTPCodecTypeVideo, clockRate: 90000, logger: logger.GetLogger()} + b.bucket = bucket.NewBucket[uint64, uint16](256, bucket.RTPMaxPktSize, bucket.RTPSeqNumOffset) + addBucketPacket(t, b, 100, 1000) + b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, true)) // marking is a no-op while disabled, but set fields anyway + + _, ok := b.GetVideoFrameCache() + require.False(t, ok) + + // audio buffers never enable the cache + a := &BufferBase{codecType: webrtc.RTPCodecTypeAudio, clockRate: 48000, logger: logger.GetLogger()} + a.EnableVideoFrameCache(0) + require.False(t, a.videoFrameCacheEnabled) +} diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 0bc566332..3bdb518ba 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -93,6 +93,23 @@ func WithForwardStats(forwardStats *ForwardStats) ReceiverOpts { } } +// DefaultVideoFrameCacheMaxDuration bounds the cached GOP (and sizes the retransmit bucket) when the video +// frame cache is enabled via WithVideoFrameCache. +const DefaultVideoFrameCacheMaxDuration = 1 * time.Second + +// WithVideoFrameCache enables the video frame cache on the receiver so a newly added down track is +// bootstrapped by replaying the cached GOP instead of triggering a PLI. maxDuration bounds the +// cached video frame (<= 0 uses DefaultVideoFrameCacheMaxDuration). No-op for audio receivers. +func WithVideoFrameCache(maxDuration time.Duration) ReceiverOpts { + return func(w *WebRTCReceiver) *WebRTCReceiver { + if maxDuration <= 0 { + maxDuration = DefaultVideoFrameCacheMaxDuration + } + w.ReceiverBase.EnableVideoFrameCache(maxDuration) + return w + } +} + // NewWebRTCReceiver creates a new webrtc track receiver func NewWebRTCReceiver( receiver *webrtc.RTPReceiver, diff --git a/pkg/sfu/receiver_base.go b/pkg/sfu/receiver_base.go index d056d252a..7a47e6ab0 100644 --- a/pkg/sfu/receiver_base.go +++ b/pkg/sfu/receiver_base.go @@ -215,6 +215,16 @@ type ReceiverBase struct { videoSizes [buffer.DefaultMaxLayerSpatial + 1]codec.VideoSize onVideoSizeChanged func() + // video frame cache: when enabled, buffers retain the current frame cache and a newly added down track is + // bootstrapped by replaying it (see EnableVideoFrameCache / replayVideoFrameCache) + videoFrameCacheEnabled bool + videoFrameCacheMaxDuration time.Duration + + // counts of subscribers bootstrapped from the video frame cache (hit) vs. those that fell back to a PLI + // because no usable video frame was cached (miss); see replayVideoFrameCache + videoFrameCacheHitCount atomic.Uint32 + videoFrameCacheMissCount atomic.Uint32 + rtt uint32 streamTrackerManager *StreamTrackerManager @@ -560,11 +570,166 @@ func (r *ReceiverBase) AddDownTrack(track TrackSender) error { track.UpTrackMaxPublishedLayerChange(r.streamTrackerManager.GetMaxPublishedLayer()) track.UpTrackMaxTemporalLayerSeenChange(r.streamTrackerManager.GetMaxTemporalLayerSeen()) + if r.videoFrameCacheEnabled && r.Kind() == webrtc.RTPCodecTypeVideo { + // Bootstrap the new track from the cached video frame before joining the live broadcast: replay the + // cached video frame (paced) and catch up to the live forwarding point, then Store so the live feed takes + // over seamlessly. Held out of the broadcast during replay to avoid interleaving replayed and + // live packets. + go func() { + r.replayVideoFrameCache(track) + if r.IsClosed() || track.IsClosed() { + return + } + r.downTrackSpreader.Store(track) + + r.params.Logger.Debugw( + "downtrack added after cached video frame replay", + "subscriberID", track.SubscriberID(), + "videoFrameCacheHitCount", r.videoFrameCacheHitCount.Load(), + "videoFrameCacheMissCount", r.videoFrameCacheMissCount.Load(), + "pliForwardedCount", r.pliForwardedCount(), + ) + }() + return nil + } + r.downTrackSpreader.Store(track) r.params.Logger.Debugw("downtrack added", "subscriberID", track.SubscriberID()) return nil } +// EnableVideoFrameCache turns on the video frame cache for this receiver: each (current and future) buffer +// retains the current cached frame, and a newly added down track is bootstrapped from it (see replayVideoFrameCache). +// maxDuration bounds the cached video frame and sizes the retransmit bucket accordingly. No-op for audio. +func (r *ReceiverBase) EnableVideoFrameCache(maxDuration time.Duration) { + if r.Kind() != webrtc.RTPCodecTypeVideo { + return + } + + r.bufferMu.Lock() + r.videoFrameCacheEnabled = true + r.videoFrameCacheMaxDuration = maxDuration + buffers := r.buffers + r.bufferMu.Unlock() + + for _, buff := range buffers { + if buff != nil { + buff.EnableVideoFrameCache(maxDuration) + } + } +} + +// videoFrameCacheReplayMaxCatchupRounds bounds how many catch-up passes a video frame cache replay makes before giving up and +// joining the live feed (replaying at ~2x real time converges quickly; this guards against a +// pathological case where live keeps outrunning the replay). +const videoFrameCacheReplayMaxCatchupRounds = 8 + +// replayVideoFrameCache bootstraps a freshly added down track by replaying the publisher's cached GOP, then the +// packets accumulated since, until it catches up to the live forwarding point. Packets are paced at +// half the (estimated) video frame interval so the replay runs at ~2x real time and converges on +// live. It writes directly to the track (which is not yet in the live broadcast), so there is no +// interleaving with live packets. +// +// It replays the highest spatial layer that currently has a cached GOP: a subscriber requesting full +// quality has its forwarder targeting the top layer, and lower layers may not even be flowing (e.g. +// paused by dynacast), so layer 0 often has no GOP. +func (r *ReceiverBase) replayVideoFrameCache(track TrackSender) { + var ( + buff buffer.BufferProvider + pkts []*buffer.ExtPacket + layer = buffer.InvalidLayerSpatial + ) + for l := int32(buffer.DefaultMaxLayerSpatial); l >= 0; l-- { + b, _ := r.getBuffer(l) + if b == nil { + continue + } + if got, ok := b.GetVideoFrameCache(); ok && len(got) > 0 { + buff, pkts, layer = b, got, l + break + } + } + + if layer == buffer.InvalidLayerSpatial { + // no usable GOP cached on any layer - the down track falls back to requesting a key frame (PLI) + missCount := r.videoFrameCacheMissCount.Inc() + r.params.Logger.Debugw( + "subscriber bootstrap: video frame cache miss, falling back to PLI", + "subscriberID", track.SubscriberID(), + "videoFrameCacheEnabled", r.videoFrameCacheEnabled, + "videoFrameCacheHitCount", r.videoFrameCacheHitCount.Load(), + "videoFrameCacheMissCount", missCount, + ) + return + } + + hitCount := r.videoFrameCacheHitCount.Inc() + half := r.videoFrameCacheReplayFrameInterval(layer) / 2 + r.params.Logger.Debugw( + "subscriber bootstrap: video frame cache hit, replaying", + "subscriberID", track.SubscriberID(), + "layer", layer, + "packets", len(pkts), + "frameIntervalHalf", half, + "videoFrameCacheHitCount", hitCount, + "videoFrameCacheMissCount", r.videoFrameCacheMissCount.Load(), + ) + + var lastSN uint64 + write := func(eps []*buffer.ExtPacket) bool { + var lastTS uint64 + for i, ep := range eps { + if r.IsClosed() || track.IsClosed() { + return false + } + // pace at frame boundaries (a new timestamp starts a new frame) + if i > 0 && ep.ExtTimestamp != lastTS && half > 0 { + time.Sleep(half) + } + lastTS = ep.ExtTimestamp + rp := *ep + rp.Arrival = mono.UnixNano() + track.WriteRTP(&rp, layer) + lastSN = ep.ExtSequenceNumber + } + return true + } + + if !write(pkts) { + return + } + + // catch up to live: replay the packets that arrived during the replay, repeating (each round is + // shorter since the replay outruns live) until nothing newer remains. + for round := 0; round < videoFrameCacheReplayMaxCatchupRounds; round++ { + if r.IsClosed() || track.IsClosed() { + return + } + more, ok := buff.GetPacketsAfter(lastSN) + if !ok || len(more) == 0 { + break + } + if !write(more) { + return + } + } +} + +// videoFrameCacheReplayFrameInterval estimates the video frame interval for the spatial layer, used to pace a +// video frame cache replay. Falls back to 30fps when the frame rate is not yet known. +func (r *ReceiverBase) videoFrameCacheReplayFrameInterval(layer int32) time.Duration { + var fps float32 + for _, f := range r.GetTemporalLayerFpsForSpatial(layer) { + if f > fps { + fps = f + } + } + if fps <= 0 { + fps = 30 + } + return time.Duration(float64(time.Second) / float64(fps)) +} + func (r *ReceiverBase) DeleteDownTrack(subscriberID livekit.ParticipantID) { r.downTrackSpreader.Free(subscriberID) r.params.Logger.Debugw("downtrack deleted", "subscriberID", subscriberID) @@ -679,6 +844,21 @@ func (r *ReceiverBase) SendPLI(layer int32, force bool) { buff.SendPLI(force) } +// pliForwardedCount returns the total number of PLIs actually forwarded to the publisher across all +// spatial-layer buffers (past each buffer's throttle gate). +func (r *ReceiverBase) pliForwardedCount() uint32 { + r.bufferMu.RLock() + defer r.bufferMu.RUnlock() + + var total uint32 + for _, buff := range r.buffers { + if buff != nil { + total += buff.PLIForwardedCount() + } + } + return total +} + func (r *ReceiverBase) getBuffer(layer int32) (buffer.BufferProvider, int32) { r.bufferMu.RLock() defer r.bufferMu.RUnlock() @@ -790,6 +970,9 @@ func (r *ReceiverBase) setupBuffer(buff buffer.BufferProvider, layer int32, rtt if r.Kind() == webrtc.RTPCodecTypeVideo && layer == 0 { buff.OnCodecChange(r.handleCodecChange) } + if r.videoFrameCacheEnabled && r.Kind() == webrtc.RTPCodecTypeVideo { + buff.EnableVideoFrameCache(r.videoFrameCacheMaxDuration) + } buff.OnStreamRestart(func(reason string) { r.restartInternal(reason, true) })