From a87232f42a83868721912b714bcff93a7439d0d1 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Fri, 28 Jul 2023 14:40:33 +0800 Subject: [PATCH] Frame integrity check for svc codec (#1914) * Frame integrity check for svc codec * Fix test * Spell --- pkg/sfu/buffer/dependencydescriptorparser.go | 19 +- pkg/sfu/buffer/frameintegrity.go | 211 ++++++++++++++++++ pkg/sfu/buffer/frameintegrity_test.go | 72 ++++++ .../dependencydescriptor.go | 45 +--- .../dependencydescriptor_test.go | 4 + pkg/sfu/videolayerselector/framechain.go | 5 + 6 files changed, 317 insertions(+), 39 deletions(-) create mode 100644 pkg/sfu/buffer/frameintegrity.go create mode 100644 pkg/sfu/buffer/frameintegrity_test.go diff --git a/pkg/sfu/buffer/dependencydescriptorparser.go b/pkg/sfu/buffer/dependencydescriptorparser.go index 17ed0732d..a3a2be794 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser.go +++ b/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 { diff --git a/pkg/sfu/buffer/frameintegrity.go b/pkg/sfu/buffer/frameintegrity.go new file mode 100644 index 000000000..1b712652e --- /dev/null +++ b/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() +} diff --git a/pkg/sfu/buffer/frameintegrity_test.go b/pkg/sfu/buffer/frameintegrity_test.go new file mode 100644 index 000000000..3e505efdd --- /dev/null +++ b/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)) + } +} diff --git a/pkg/sfu/videolayerselector/dependencydescriptor.go b/pkg/sfu/videolayerselector/dependencydescriptor.go index ef63f6526..555437c6d 100644 --- a/pkg/sfu/videolayerselector/dependencydescriptor.go +++ b/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 diff --git a/pkg/sfu/videolayerselector/dependencydescriptor_test.go b/pkg/sfu/videolayerselector/dependencydescriptor_test.go index 8b4aad771..e9e158fda 100644 --- a/pkg/sfu/videolayerselector/dependencydescriptor_test.go +++ b/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{ diff --git a/pkg/sfu/videolayerselector/framechain.go b/pkg/sfu/videolayerselector/framechain.go index 6dc460c28..5edddfad9 100644 --- a/pkg/sfu/videolayerselector/framechain.go +++ b/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]