diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 8c0d87977..d2f156e5d 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -341,6 +341,7 @@ type DownTrack struct { keyFrameRequesterChMu sync.RWMutex keyFrameRequesterCh chan struct{} keyFrameRequesterChClosed bool + keyFrameRequesterWakeups atomic.Uint32 retransmitChMu sync.RWMutex retransmitCh chan struct{} @@ -956,18 +957,20 @@ func (d *DownTrack) keyFrameRequester() { defer timer.Stop() for !d.IsClosed() { - timer.Reset(getInterval()) - select { case _, more := <-d.keyFrameRequesterCh: if !more { return } if !timer.Stop() { - <-timer.C + select { + case <-timer.C: + default: + } } case <-timer.C: } + d.keyFrameRequesterWakeups.Inc() locked, layer := d.forwarder.CheckSync() if !locked && layer != buffer.InvalidLayerSpatial && d.writable.Load() { @@ -984,6 +987,13 @@ func (d *DownTrack) keyFrameRequester() { sal.OnAvailableLayersChanged(d) } } + + // Re-arm only while there is something to wait for: a requested layer that is not locked yet, + // or a lock state that can change without an event. Once locked, sleep until the + // next event so a locked track does not wake up every interval. + if (!locked && layer != buffer.InvalidLayerSpatial) || d.forwarder.NeedsPeriodicSyncCheck() { + timer.Reset(getInterval()) + } } } @@ -1197,6 +1207,8 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { if tp.isSwitching { d.postMaxLayerNotifierEvent("switching") + // the current layer moved on the packet path, so the layer lock may have changed + d.postKeyFrameRequestEvent() } if tp.isResuming { @@ -1443,6 +1455,8 @@ func (d *DownTrack) handleMute(muted bool, changed bool) { // and that could turn on/off layers on publisher side. // d.postMaxLayerNotifierEvent("mute") + // mute resyncs the forwarder, so the layer lock has to be re-acquired on unmute + d.postKeyFrameRequestEvent() if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnSubscriptionChanged(d) @@ -1659,6 +1673,8 @@ func (d *DownTrack) UpTrackBitrateAvailabilityChange() { func (d *DownTrack) UpTrackMaxPublishedLayerChange(maxPublishedLayer int32) { if d.forwarder.SetMaxPublishedLayer(maxPublishedLayer) { + // may arm the acquisition grace, which the key frame requester has to expire + d.postKeyFrameRequestEvent() if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnMaxPublishedSpatialChanged(d) } @@ -1804,6 +1820,7 @@ func (d *DownTrack) Pause() VideoAllocation { func (d *DownTrack) Resync() { d.forwarder.Resync() d.flushSequencer() + d.postKeyFrameRequestEvent() } // flushSequencer discards recorded packet metadata on a stream restart so that NACK diff --git a/pkg/sfu/downtrack_downstream_integration_test.go b/pkg/sfu/downtrack_downstream_integration_test.go index 1f946a531..41282e46c 100644 --- a/pkg/sfu/downtrack_downstream_integration_test.go +++ b/pkg/sfu/downtrack_downstream_integration_test.go @@ -919,3 +919,67 @@ func TestDownTrackSendsProbePackets(t *testing.T) { }) } } + +// ----------------------------------------------------------------------------- +// keyFrameRequester: PLI only while a layer lock is pending +// ----------------------------------------------------------------------------- + +// TestDownTrackKeyFrameRequesterArmsOnlyWhilePending verifies that the key frame +// requester keeps sending PLIs while the target layer is not locked, and goes quiet +// once a key frame on the target layer locks it (no periodic wake-ups while locked). +func TestDownTrackKeyFrameRequesterArmsOnlyWhilePending(t *testing.T) { + h := vnettest.NewHosts(t) + factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() + p := pacer.NewPassThrough(logger.GetLogger(), newNullBWE()) + t.Cleanup(p.Stop) + + dh := newBoundDownTrack(t, h, factory, vp8CodecParams, p, vnettest.MediaEngineConfig{Video: true}) + + // allocate a target layer (spatial 0) without any media, so the lock is pending + var brs Bitrates + brs[0][0] = 100_000 + dh.receiver.GetLayeredBitrateReturns([]int32{0}, brs) + dh.dt.UpTrackMaxPublishedLayerChange(0) + dh.dt.SetMaxSpatialLayer(0) + alloc := dh.dt.AllocateOptimal(false, false) + require.Equal(t, int32(0), alloc.TargetLayer.Spatial, "expected target layer to be allocated") + + // unlocked: PLIs must keep coming at the key frame interval + require.Eventually(t, func() bool { + return dh.receiver.SendPLICallCount() >= 3 + }, 5*time.Second, 20*time.Millisecond, "expected repeated PLIs while layer lock is pending") + + // a key frame on the target layer locks it + vp8 := &codec.VP8{ + FirstByte: 0x10, + S: true, + PictureID: 1, + IsKeyFrame: true, + } + ep, err := testutils.GetTestExtPacketVP8(&testutils.TestExtPacketParams{ + SequenceNumber: 5000, + Timestamp: 180000, + SSRC: 0x33333333, + PayloadType: 96, + PayloadSize: 50, + IsKeyFrame: true, + Marker: true, + }, vp8) + require.NoError(t, err) + ep.Packet.Payload = distinctivePayload(0x40, 50) + require.Equal(t, int32(1), dh.dt.WriteRTP(ep, 0), "keyframe should forward") + + // let an already armed interval fire, then the requester must stay asleep + time.Sleep(2 * KeyFrameIntervalMaxForTest) + before := dh.receiver.SendPLICallCount() + wakeups := dh.dt.KeyFrameRequesterWakeupsForTest() + time.Sleep(2 * KeyFrameIntervalMaxForTest) + require.Equal(t, wakeups, dh.dt.KeyFrameRequesterWakeupsForTest(), "key frame requester must not wake up once the layer is locked") + require.Equal(t, before, dh.receiver.SendPLICallCount(), "no PLI expected once the layer is locked") + + // a resync drops the lock and must wake the requester again + dh.dt.Resync() + require.Eventually(t, func() bool { + return dh.receiver.SendPLICallCount() > before + }, 5*time.Second, 20*time.Millisecond, "expected PLI after resync") +} diff --git a/pkg/sfu/export_test.go b/pkg/sfu/export_test.go index 449289e20..b6536362a 100644 --- a/pkg/sfu/export_test.go +++ b/pkg/sfu/export_test.go @@ -14,7 +14,11 @@ package sfu -import "github.com/livekit/livekit-server/pkg/sfu/buffer" +import ( + "time" + + "github.com/livekit/livekit-server/pkg/sfu/buffer" +) // Test-only seams that expose DownTrack internals to the external sfu_test package. // This file is only compiled into the test binary (its name ends in _test.go), so it @@ -71,3 +75,9 @@ func (d *DownTrack) RetransmitForTest( } return d.retransmitPacket(&epm, sourcePkt, false) } + +const KeyFrameIntervalMaxForTest = keyFrameIntervalMax * time.Millisecond + +func (d *DownTrack) KeyFrameRequesterWakeupsForTest() uint32 { + return d.keyFrameRequesterWakeups.Load() +} diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index c546796fa..f69e6fec5 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -360,6 +360,21 @@ func (f *Forwarder) MaybeExpireAcquireGrace() bool { return !f.vls.GetCurrent().IsValid() } +// NeedsPeriodicSyncCheck reports whether the layer lock state can change without an event, so the +// key frame requester has to keep polling it: an initial-acquisition grace deadline is armed +// (MaybeExpireAcquireGrace may still fire), or layers are selected with the dependency descriptor, +// which can lose the lock on the packet path (key frame mismatch, broken frame chain). +func (f *Forwarder) NeedsPeriodicSyncCheck() bool { + f.lock.RLock() + defer f.lock.RUnlock() + + if f.acquireDeadline != 0 { + return true + } + _, isDD := f.vls.(*videolayerselector.DependencyDescriptor) + return isDD +} + func (f *Forwarder) SetMaxTemporalLayerSeen(maxTemporalLayerSeen int32) bool { f.lock.Lock() defer f.lock.Unlock() diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index cc7f92f0d..0ee950ae6 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -2502,3 +2502,22 @@ func TestPacerScratchFitsInlineDependencyDescriptor(t *testing.T) { require.Len(t, p.HoldExtension(make([]byte, dd.MaxInlineExtensionSize)), dd.MaxInlineExtensionSize) require.Nil(t, p.HoldExtension(make([]byte, dd.MaxInlineExtensionSize+1))) } + +func TestForwarderNeedsPeriodicSyncCheck(t *testing.T) { + f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + require.False(t, f.NeedsPeriodicSyncCheck()) + + // VP8 with the dependency descriptor uses it only for temporal layers + f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) + require.False(t, f.NeedsPeriodicSyncCheck()) + + // a new max published layer before any layer is forwarded arms the acquisition grace + require.True(t, f.SetMaxPublishedLayer(1)) + require.True(t, f.NeedsPeriodicSyncCheck()) + + // spatial layers selected with the dependency descriptor + f = newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + av1 := webrtc.RTPCodecCapability{MimeType: mime.MimeTypeAV1.String(), ClockRate: 90000} + f.DetermineCodec(av1, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) + require.True(t, f.NeedsPeriodicSyncCheck()) +}