diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index d15fe5cdd..a413ad584 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -1787,6 +1787,9 @@ func (f *Forwarder) Resync() { func (f *Forwarder) resyncLocked() { f.vls.SetCurrent(buffer.InvalidLayer) f.lastSSRC = 0 + if f.ddForwarder != nil { + f.ddForwarder.Resync() + } if f.pubMuted { f.resumeBehindThreshold = ResumeBehindThresholdSeconds } diff --git a/pkg/sfu/videolayerselector/ddforwarder.go b/pkg/sfu/videolayerselector/ddforwarder.go index 2080f34d7..5c6fb8391 100644 --- a/pkg/sfu/videolayerselector/ddforwarder.go +++ b/pkg/sfu/videolayerselector/ddforwarder.go @@ -30,10 +30,17 @@ import ( // frame numbers of a new layer are close to those of the old one. The numbers sent to the // subscriber must still never go back: a key frame older than the last structure the // subscriber received is dropped. So whenever the stream changes, a layer switch, a stream -// restart or a new forwarder, the numbers are moved up past the last one sent if needed. +// restart, a resync or a new forwarder, the numbers are moved up past the last one sent if +// needed. They must not jump too far ahead either: the subscriber unwraps the 16 bit frame +// number, so a jump of half its range or more reads as a jump back. The publisher keeps +// counting while a muted subscriber gets nothing, so a long mute would cause such a jump. // Otherwise they pass through with the current offset, so frame diffs and chain diffs stay // valid. A migrated subscriber continues with the last number sent, which is part of the // migrated forwarder state. +// maxFrameNumJump is the largest jump in the frame numbers sent on a stream change, well below +// half the range of the 16 bit frame number. +const maxFrameNumJump = 1 << 14 + type DDForwarder struct { logger logger.Logger @@ -159,8 +166,18 @@ func (d *DDForwarder) Forward(extPkt *buffer.ExtPacket, temporal int32, result * } // rebase moves the frame numbers up when a new stream would go back past the last one sent +// Resync starts the stream again on the next key frame with a structure, like a new forwarder, +// keeping the last number sent. The forwarder resyncs on a mute, for example. +func (d *DDForwarder) Resync() { + d.initialized = false +} + func (d *DDForwarder) rebase(extFrameNum uint64) { - if d.hasLast && extFrameNum+d.frameNumOffset <= d.extLastFrameNum { + if !d.hasLast { + return + } + extOutFrameNum := extFrameNum + d.frameNumOffset + if extOutFrameNum <= d.extLastFrameNum || extOutFrameNum-d.extLastFrameNum > maxFrameNumJump { d.frameNumOffset = d.extLastFrameNum + 1 - extFrameNum } } diff --git a/pkg/sfu/videolayerselector/ddforwarder_test.go b/pkg/sfu/videolayerselector/ddforwarder_test.go index 701fe2503..66a155e7f 100644 --- a/pkg/sfu/videolayerselector/ddforwarder_test.go +++ b/pkg/sfu/videolayerselector/ddforwarder_test.go @@ -183,6 +183,24 @@ func TestDDForwarderRestartOnDeltaFrame(t *testing.T) { require.Equal(t, uint16(1004), d.forward(f, d.packet(1, 5, false, 1), 0).FrameNumber) } +func TestDDForwarderResync(t *testing.T) { + d := newDDForwarderTest(t) + f := NewDDForwarder(logger.GetLogger()) + require.Equal(t, uint16(1000), d.forward(f, d.packet(1, 1000, true, 0), 0).FrameNumber) + require.Equal(t, uint16(1001), d.forward(f, d.packet(1, 1001, false, 0), 0).FrameNumber) + + // a long mute: the publisher counted 40000 frames while nothing was forwarded, the + // subscriber would take a jump of more than half the 16 bit range as a jump back + f.Resync() + require.Nil(t, d.forward(f, d.packet(1, 41001, false, 0), 0)) + require.Equal(t, uint16(1002), d.forward(f, d.packet(1, 41002, true, 0), 0).FrameNumber) + require.Equal(t, uint16(1003), d.forward(f, d.packet(1, 41003, false, 0), 0).FrameNumber) + + // a short mute keeps the current offset, the gap stays + f.Resync() + require.Equal(t, uint16(1103), d.forward(f, d.packet(1, 41103, true, 0), 0).FrameNumber) +} + func TestDDForwarderPublisherActiveDecodeTargets(t *testing.T) { d := newDDForwarderTest(t) f := NewDDForwarder(logger.GetLogger())