Frame integrity check for svc codec (#1914)

* Frame integrity check for svc codec

* Fix test

* Spell

pkg/sfu/buffer/dependencydescriptorparser.go

@@ -33,10 +33,12 @@ type DependencyDescriptorParser struct {
 	onMaxLayerChanged func(int32, int32)
 	decodeTargets     []DependencyDescriptorDecodeTarget
 
-	wrapAround                *utils.WrapAround[uint16, uint64]
+	seqWrapAround             *utils.WrapAround[uint16, uint64]
+	frameWrapAround           *utils.WrapAround[uint16, uint64]
 	structureExtSeq           uint64
 	activeDecodeTargetsExtSeq uint64
 	activeDecodeTargetsMask   uint32
+	frameChecker              *FrameIntegrityChecker
 }
 
 func NewDependencyDescriptorParser(ddExtID uint8, logger logger.Logger, onMaxLayerChanged func(int32, int32)) *DependencyDescriptorParser {
@@ -45,7 +47,9 @@ func NewDependencyDescriptorParser(ddExtID uint8, logger logger.Logger, onMaxLay
 		ddExtID:           ddExtID,
 		logger:            logger,
 		onMaxLayerChanged: onMaxLayerChanged,
-		wrapAround:        utils.NewWrapAround[uint16, uint64](),
+		seqWrapAround:     utils.NewWrapAround[uint16, uint64](),
+		frameWrapAround:   utils.NewWrapAround[uint16, uint64](),
+		frameChecker:      NewFrameIntegrityChecker(180, 1024), // 2seconds for L3T3 30fps video
 	}
 }
 
@@ -55,6 +59,8 @@ type ExtDependencyDescriptor struct {
 	DecodeTargets              []DependencyDescriptorDecodeTarget
 	StructureUpdated           bool
 	ActiveDecodeTargetsUpdated bool
+	Integrity                  bool
+	ExtFrameNum                uint64
 }
 
 func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescriptor, VideoLayer, error) {
@@ -75,14 +81,19 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr
 		return nil, videoLayer, err
 	}
 
-	extSeq := r.wrapAround.Update(pkt.SequenceNumber).ExtendedVal
+	extSeq := r.seqWrapAround.Update(pkt.SequenceNumber).ExtendedVal
 
 	if ddVal.FrameDependencies != nil {
 		videoLayer.Spatial, videoLayer.Temporal = int32(ddVal.FrameDependencies.SpatialId), int32(ddVal.FrameDependencies.TemporalId)
 	}
 
+	extFN := r.frameWrapAround.Update(ddVal.FrameNumber).ExtendedVal
+	r.frameChecker.AddPacket(extSeq, extFN, &ddVal)
+
 	extDD := &ExtDependencyDescriptor{
-		Descriptor: &ddVal,
+		Descriptor:  &ddVal,
+		ExtFrameNum: extFN,
+		Integrity:   r.frameChecker.FrameIntegrity(extFN),
 	}
 
 	if ddVal.AttachedStructure != nil {

pkg/sfu/buffer/frameintegrity.go

@@ -0,0 +1,211 @@
+package buffer
+
+import (
+	dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
+)
+
+type FrameEntity struct {
+	startSeq  *uint64
+	endSeq    *uint64
+	integrity bool
+
+	packetsConsective func(uint64, uint64) bool
+}
+
+func (fe *FrameEntity) AddPacket(extSeq uint64, ddVal *dd.DependencyDescriptor) {
+	// duplicate packet
+	if fe.integrity {
+		return
+	}
+
+	if fe.startSeq == nil && ddVal.FirstPacketInFrame {
+		fe.startSeq = &extSeq
+	}
+	if fe.endSeq == nil && ddVal.LastPacketInFrame {
+		fe.endSeq = &extSeq
+	}
+
+	if fe.startSeq != nil && fe.endSeq != nil {
+		if fe.packetsConsective(*fe.startSeq, *fe.endSeq) {
+			fe.integrity = true
+		}
+	}
+}
+
+func (fe *FrameEntity) Reset() {
+	fe.integrity = false
+	fe.startSeq, fe.endSeq = nil, nil
+}
+
+func (fe *FrameEntity) Integrity() bool {
+	return fe.integrity
+}
+
+// ------------------------------
+
+type PacketHistory struct {
+	base        uint64
+	last        uint64
+	bits        []uint64
+	packetCount int
+	inited      bool
+}
+
+func NewPacketHistory(packetCount int) *PacketHistory {
+	packetCount = (packetCount + 63) / 64 * 64
+	return &PacketHistory{
+		bits:        make([]uint64, packetCount/64),
+		packetCount: packetCount,
+	}
+}
+
+func (ph *PacketHistory) AddPacket(extSeq uint64) {
+	if !ph.inited {
+		ph.inited = true
+		ph.base = uint64(extSeq)
+		// set base to extSeq-100 to avoid out-of-order packets belongs to first frame to be dropped
+		if ph.base > 100 {
+			ph.base -= 100
+		} else {
+			ph.base = 0
+		}
+		ph.last = uint64(extSeq)
+		ph.set(extSeq, true)
+		return
+	}
+
+	if extSeq <= ph.base {
+		// too old
+		return
+	}
+
+	if extSeq <= ph.last {
+		if ph.last-extSeq < uint64(ph.packetCount) {
+			ph.set(extSeq, true)
+		}
+		return
+	}
+
+	for i := ph.last + 1; i < extSeq; i++ {
+		ph.set(i, false)
+	}
+
+	ph.set(extSeq, true)
+	ph.last = extSeq
+}
+
+func (ph *PacketHistory) getPos(seq uint64) (index, offset int) {
+	idx := (seq - ph.base) % uint64(ph.packetCount)
+	return int(idx >> 6), int(idx % 64)
+}
+
+func (ph *PacketHistory) set(seq uint64, received bool) {
+	idx, offset := ph.getPos(seq)
+	if !received {
+		ph.bits[idx] &= ^(1 << offset)
+	} else {
+		ph.bits[idx] |= 1 << (offset)
+	}
+}
+
+func (ph *PacketHistory) PacketsConsecutive(start, end uint64) bool {
+	if start > end {
+		return false
+	}
+
+	if end-start >= uint64(ph.packetCount) {
+		return false
+	}
+
+	startIndex, startOffset := ph.getPos(start)
+	endIndex, endOffset := ph.getPos(end)
+
+	if startIndex == endIndex && end-start <= 64 {
+		testBits := uint64((1<<(endOffset-startOffset+1))-1) << startOffset
+		return ph.bits[startIndex]&testBits == testBits
+	}
+
+	if (ph.bits[startIndex]>>(startOffset))+1 != 1<<(64-startOffset) {
+		return false
+	}
+
+	for i := startIndex + 1; i != endIndex; i++ {
+		if i == len(ph.bits) {
+			i = 0
+			if i == endIndex {
+				break
+			}
+		}
+		if ph.bits[i]+1 != 0 {
+			return false
+		}
+	}
+
+	testBits := uint64((1 << (endOffset + 1)) - 1)
+	return ph.bits[endIndex]&testBits == testBits
+}
+
+// ------------------------------
+
+type FrameIntegrityChecker struct {
+	frameCount int
+	frames     []FrameEntity
+	base       uint64
+	last       uint64
+
+	pktHistory *PacketHistory
+	inited     bool
+}
+
+func NewFrameIntegrityChecker(frameCount, packetCount int) *FrameIntegrityChecker {
+	fc := &FrameIntegrityChecker{
+		frames:     make([]FrameEntity, frameCount),
+		pktHistory: NewPacketHistory(packetCount),
+		frameCount: frameCount,
+	}
+
+	for i := range fc.frames {
+		fc.frames[i].packetsConsective = fc.pktHistory.PacketsConsecutive
+		fc.frames[i].Reset()
+	}
+	return fc
+}
+
+func (fc *FrameIntegrityChecker) AddPacket(extSeq uint64, extFrameNum uint64, ddVal *dd.DependencyDescriptor) {
+	fc.pktHistory.AddPacket(extSeq)
+
+	if !fc.inited {
+		fc.inited = true
+		fc.base = extFrameNum
+		fc.last = extFrameNum
+	}
+
+	if extFrameNum < fc.base {
+		// frame too old
+		return
+	}
+
+	if extFrameNum <= fc.last {
+		if fc.last-extFrameNum >= uint64(fc.frameCount) {
+			// frame too old
+			return
+		}
+		fc.frames[int(extFrameNum-fc.base)%fc.frameCount].AddPacket(extSeq, ddVal)
+		return
+	}
+
+	// reset missing frames
+	for i := fc.last + 1; i <= extFrameNum; i++ {
+		fc.frames[int(i-fc.base)%fc.frameCount].Reset()
+	}
+	fc.frames[int(extFrameNum-fc.base)%fc.frameCount].AddPacket(extSeq, ddVal)
+	fc.last = extFrameNum
+}
+
+func (fc *FrameIntegrityChecker) FrameIntegrity(extFrameNum uint64) bool {
+	if extFrameNum < fc.base || extFrameNum > fc.last || fc.last-extFrameNum >= uint64(fc.frameCount) {
+		return false
+	}
+
+	return fc.frames[int(extFrameNum-fc.base)%fc.frameCount].Integrity()
+}

pkg/sfu/buffer/frameintegrity_test.go

@@ -0,0 +1,72 @@
+package buffer
+
+import (
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
+)
+
+func TestFrameIntegrityChecker(t *testing.T) {
+	fc := NewFrameIntegrityChecker(100, 1000)
+
+	// first frame out of order
+	fc.AddPacket(10, 10, &dependencydescriptor.DependencyDescriptor{})
+	require.False(t, fc.FrameIntegrity(10))
+	fc.AddPacket(9, 10, &dependencydescriptor.DependencyDescriptor{FirstPacketInFrame: true})
+	require.False(t, fc.FrameIntegrity(10))
+	fc.AddPacket(11, 10, &dependencydescriptor.DependencyDescriptor{LastPacketInFrame: true})
+	require.True(t, fc.FrameIntegrity(10))
+
+	// single packet frame
+	fc.AddPacket(100, 100, &dependencydescriptor.DependencyDescriptor{FirstPacketInFrame: true, LastPacketInFrame: true})
+	require.True(t, fc.FrameIntegrity(100))
+	require.False(t, fc.FrameIntegrity(101))
+	require.False(t, fc.FrameIntegrity(99))
+
+	// frame too old than first frame
+	fc.AddPacket(99, 99, &dependencydescriptor.DependencyDescriptor{FirstPacketInFrame: true, LastPacketInFrame: true})
+
+	// multiple packet frame, out of order
+	fc.AddPacket(2001, 2001, &dependencydescriptor.DependencyDescriptor{})
+	require.False(t, fc.FrameIntegrity(2001))
+	require.False(t, fc.FrameIntegrity(1999))
+	// out of frame count(100)
+	require.False(t, fc.FrameIntegrity(100))
+	require.False(t, fc.FrameIntegrity(1900))
+
+	fc.AddPacket(2000, 2001, &dependencydescriptor.DependencyDescriptor{FirstPacketInFrame: true})
+	require.False(t, fc.FrameIntegrity(2001))
+	fc.AddPacket(2002, 2001, &dependencydescriptor.DependencyDescriptor{LastPacketInFrame: true})
+	require.True(t, fc.FrameIntegrity(2001))
+	// duplicate packet
+	fc.AddPacket(2001, 2001, &dependencydescriptor.DependencyDescriptor{})
+	require.True(t, fc.FrameIntegrity(2001))
+
+	// frame too old
+	fc.AddPacket(900, 1900, &dependencydescriptor.DependencyDescriptor{FirstPacketInFrame: true, LastPacketInFrame: true})
+	require.False(t, fc.FrameIntegrity(1900))
+
+	for frame := uint64(2002); frame < 2102; frame++ {
+		// large frame (1000 packets) out of order / retransmitted
+		firstFrame := uint64(3000 + (frame-2002)*1000)
+		lastFrame := uint64(3999 + (frame-2002)*1000)
+		frames := make([]uint64, 0, lastFrame-firstFrame+1)
+		for i := firstFrame; i <= lastFrame; i++ {
+			frames = append(frames, i)
+		}
+		require.False(t, fc.FrameIntegrity(frame))
+		rand.Seed(int64(frame))
+		rand.Shuffle(len(frames), func(i, j int) { frames[i], frames[j] = frames[j], frames[i] })
+		for i, f := range frames {
+			fc.AddPacket(f, frame, &dependencydescriptor.DependencyDescriptor{
+				FirstPacketInFrame: f == firstFrame,
+				LastPacketInFrame:  f == lastFrame,
+			})
+			require.Equal(t, i == len(frames)-1, fc.FrameIntegrity(frame), i)
+		}
+		require.True(t, fc.FrameIntegrity(frame))
+	}
+}

pkg/sfu/videolayerselector/dependencydescriptor.go

@@ -20,14 +20,12 @@ import (
 
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	dede "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
-	"github.com/livekit/livekit-server/pkg/sfu/utils"
 	"github.com/livekit/protocol/logger"
 )
 
 type DependencyDescriptor struct {
 	*Base
 
-	frameNum  *utils.WrapAround[uint16, uint64]
 	decisions *SelectorDecisionCache
 
 	previousActiveDecodeTargetsBitmask *uint32
@@ -43,7 +41,6 @@ type DependencyDescriptor struct {
 func NewDependencyDescriptor(logger logger.Logger) *DependencyDescriptor {
 	return &DependencyDescriptor{
 		Base:      NewBase(logger),
-		frameNum:  utils.NewWrapAround[uint16, uint64](),
 		decisions: NewSelectorDecisionCache(256, 80),
 	}
 }
@@ -51,7 +48,6 @@ func NewDependencyDescriptor(logger logger.Logger) *DependencyDescriptor {
 func NewDependencyDescriptorFromNull(vls VideoLayerSelector) *DependencyDescriptor {
 	return &DependencyDescriptor{
 		Base:      vls.(*Null).Base,
-		frameNum:  utils.NewWrapAround[uint16, uint64](),
 		decisions: NewSelectorDecisionCache(256, 80),
 	}
 }
@@ -72,8 +68,7 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r
 
 	dd := ddwdt.Descriptor
 
-	frameNum := d.frameNum.Update(dd.FrameNumber)
-	extFrameNum := frameNum.ExtendedVal
+	extFrameNum := ddwdt.ExtFrameNum
 
 	fd := dd.FrameDependencies
 	incomingLayer := buffer.VideoLayer{
@@ -117,6 +112,8 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r
 	}
 	var dti dede.DecodeTargetIndication
 	d.decodeTargetsLock.RLock()
+
+	// decodeTargets be sorted from high to low, find the highest decode target that is active and integrity
 	for _, dt := range d.decodeTargets {
 		if !dt.Active() || dt.Layer.Spatial > d.targetLayer.Spatial || dt.Layer.Temporal > d.targetLayer.Temporal {
 			continue
@@ -133,28 +130,14 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r
 			return
 		}
 
-		// Keep forwarding the lower spatial with temporal layer 0 to keep the lower frame chain intact,
-		// it will cost a few extra bits as those frames might not be present in the current target
-		// but will make the subscriber switch to lower layer seamlessly without pli.
 		if frameResult.TargetValid {
-			if highestDecodeTarget.Target == -1 {
-				highestDecodeTarget = dt.DependencyDescriptorDecodeTarget
-				dti = frameResult.DTI
-			} else if dt.Layer.Spatial < highestDecodeTarget.Layer.Spatial && dt.Layer.Temporal == 0 &&
-				frameResult.DTI != dede.DecodeTargetNotPresent && frameResult.DTI != dede.DecodeTargetDiscardable {
-				dti = frameResult.DTI
-			}
+			highestDecodeTarget = dt.DependencyDescriptorDecodeTarget
+			dti = frameResult.DTI
+			break
 		}
 	}
 	d.decodeTargetsLock.RUnlock()
 
-	// DD-TODO : we don't have a rtp queue to ensure the order of packets now,
-	// so we don't know packet is lost/out of order, that cause us can't detect
-	// frame integrity, entire frame is forwareded, whether frame chain is broken.
-	// So use a simple check here, assume all the reference frame is forwarded and
-	// only check DTI of the active decode target.
-	// it is not effeciency, at last we need check frame chain integrity.
-
 	if highestDecodeTarget.Target < 0 {
 		// no active decode target, do not select
 		// d.logger.Debugw(fmt.Sprintf("drop packet for no target found, decodeTargets %v, tagetLayer %v, incoming %v",
@@ -218,6 +201,7 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r
 
 		d.previousActiveDecodeTargetsBitmask = d.activeDecodeTargetsBitmask
 		d.activeDecodeTargetsBitmask = buffer.GetActiveDecodeTargetBitmask(d.currentLayer, ddwdt.DecodeTargets)
+		d.logger.Debugw("switch to target", "highest", highestDecodeTarget.Layer, "current", d.currentLayer, "bitmask", *d.activeDecodeTargetsBitmask)
 	}
 
 	ddExtension := &dede.DependencyDescriptorExtension{
@@ -241,18 +225,9 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r
 		result.DependencyDescriptorExtension = bytes
 	}
 
-	// DD-TODO START
-	// Ideally should add this frame only on the last packet of the frame and if all packets of the frame have been selected.
-	// But, adding on any packet so that any out-of-order packets within a frame can be fowarded.
-	// But, that could result in decodability/chain integrity to erroneously pass (i. e. in the case of lost packet in this
-	// frame, this frame is not decodable and hence the chain is broken).
-	//
-	// Note that packets can get lost in the forwarded path also. That will be handled by receiver sending PLI.
-	//
-	// Within SFU, there is more work to do to ensure integrity of forwarded packets/frames to adhere to the complete design
-	// goal of dependency descriptor
-	// DD-TODO END
-	d.decisions.AddForwarded(extFrameNum)
+	if ddwdt.Integrity {
+		d.decisions.AddForwarded(extFrameNum)
+	}
 	result.RTPMarker = extPkt.Packet.Header.Marker || (dd.LastPacketInFrame && d.currentLayer.Spatial == int32(fd.SpatialId))
 	result.IsSelected = true
 	return

pkg/sfu/videolayerselector/dependencydescriptor_test.go

@@ -321,6 +321,8 @@ func createDDFrames(maxLayer buffer.VideoLayer, startFrameNumber uint16) []*buff
 			DecodeTargets:              decodeTargets,
 			StructureUpdated:           true,
 			ActiveDecodeTargetsUpdated: true,
+			Integrity:                  true,
+			ExtFrameNum:                uint64(startFrameNumber),
 		},
 		Packet: &rtp.Packet{
 			Header: rtp.Header{
@@ -370,6 +372,8 @@ func createDDFrames(maxLayer buffer.VideoLayer, startFrameNumber uint16) []*buff
 						},
 					},
 					DecodeTargets: decodeTargets,
+					Integrity:     true,
+					ExtFrameNum:   uint64(startFrameNumber),
 				},
 				Packet: &rtp.Packet{
 					Header: rtp.Header{

pkg/sfu/videolayerselector/framechain.go

@@ -87,10 +87,15 @@ func (fc *FrameChain) OnFrame(extFrameNum uint64, fd *dd.FrameDependencyTemplate
 }
 
 func (fc *FrameChain) OnExpectFrameChanged(frameNum uint64, decision selectorDecision) {
+	if fc.broken {
+		return
+	}
+
 	for i, f := range fc.expectFrames {
 		if f == frameNum {
 			if decision != selectorDecisionForwarded {
 				fc.broken = true
+				fc.logger.Debugw("frame chain broken", "chanIdx", fc.chainIdx, "sd", decision, "frame", frameNum)
 			}
 			fc.expectFrames[i] = fc.expectFrames[len(fc.expectFrames)-1]
 			fc.expectFrames = fc.expectFrames[:len(fc.expectFrames)-1]