diff --git a/.github/workflows/slack-notifier.yaml b/.github/workflows/slack-notifier.yaml new file mode 100644 index 000000000..7770345bf --- /dev/null +++ b/.github/workflows/slack-notifier.yaml @@ -0,0 +1,25 @@ +name: PR Slack Notifier + +on: + pull_request: + types: [review_requested, reopened, closed] + pull_request_review: + types: [submitted] + +permissions: + contents: read + pull-requests: write + issues: write + +concurrency: + group: pr-slack-${{ github.event.pull_request.number }}-${{ github.workflow }} + cancel-in-progress: false + +jobs: + notify-devs: + runs-on: ubuntu-latest + steps: + - uses: livekit/slack-notifier-action@main + with: + config_json: ${{ secrets.SLACK_NOTIFY_CONFIG_JSON }} + slack_token: ${{ secrets.SLACK_PR_NOTIFIER_TOKEN }} diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index feb95ee68..587b315bd 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -338,7 +338,7 @@ func (b *Buffer) createDDParserAndFrameRateCalculator() { } b.ddParser = NewDependencyDescriptorParser(b.ddExtID, b.logger, func(spatial, temporal int32) { frc.SetMaxLayer(spatial, temporal) - }) + }, false) } } diff --git a/pkg/sfu/buffer/dependencydescriptorparser.go b/pkg/sfu/buffer/dependencydescriptorparser.go index 3fbfaf9cc..c7482af80 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser.go +++ b/pkg/sfu/buffer/dependencydescriptorparser.go @@ -17,6 +17,7 @@ package buffer import ( "fmt" "sort" + "time" "github.com/pion/rtp" "go.uber.org/atomic" @@ -27,6 +28,14 @@ import ( "github.com/livekit/protocol/logger" ) +const ( + ddRestartThreshold = 30 * time.Second + + // frame integrity check 2 seconds for L3T3 30fps video + integrityCheckFrame = 180 + integrityCheckPkt = 1024 +) + var ( ErrFrameEarlierThanKeyFrame = fmt.Errorf("frame is earlier than current keyframe") ErrDDStructureAttachedToNonFirstPacket = fmt.Errorf("dependency descriptor structure is attached to non-first packet of a frame") @@ -48,16 +57,22 @@ type DependencyDescriptorParser struct { frameChecker *FrameIntegrityChecker ddNotFoundCount atomic.Uint32 + + // restart detection + restartGeneration int + enableRestart bool + lastPacketAt time.Time } -func NewDependencyDescriptorParser(ddExtID uint8, logger logger.Logger, onMaxLayerChanged func(int32, int32)) *DependencyDescriptorParser { +func NewDependencyDescriptorParser(ddExtID uint8, logger logger.Logger, onMaxLayerChanged func(int32, int32), enableRestart bool) *DependencyDescriptorParser { return &DependencyDescriptorParser{ ddExtID: ddExtID, logger: logger, onMaxLayerChanged: onMaxLayerChanged, seqWrapAround: utils.NewWrapAround[uint16, uint64](utils.WrapAroundParams{IsRestartAllowed: false}), frameWrapAround: utils.NewWrapAround[uint16, uint64](utils.WrapAroundParams{IsRestartAllowed: false}), - frameChecker: NewFrameIntegrityChecker(180, 1024), // 2seconds for L3T3 30fps video + frameChecker: NewFrameIntegrityChecker(integrityCheckFrame, integrityCheckPkt), + enableRestart: enableRestart, } } @@ -71,6 +86,10 @@ type ExtDependencyDescriptor struct { ExtFrameNum uint64 // the frame number of the keyframe which the current frame depends on ExtKeyFrameNum uint64 + + // increase when the stream restarts, clear and reinitialize all dd state includes + // attached structure, frame chain, decode target. + RestartGeneration int } func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescriptor, VideoLayer, error) { @@ -84,6 +103,16 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr return nil, videoLayer, ErrDDExtentionNotFound } + var restart bool + if r.enableRestart { + if !r.lastPacketAt.IsZero() && time.Since(r.lastPacketAt) > ddRestartThreshold { + r.restart() + restart = true + r.logger.Debugw("dependency descriptor parser restart stream", "generation", r.restartGeneration) + } + r.lastPacketAt = time.Now() + } + var ddVal dd.DependencyDescriptor ext := &dd.DependencyDescriptorExtension{ Descriptor: &ddVal, @@ -103,7 +132,8 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr videoLayer.Spatial, videoLayer.Temporal = int32(ddVal.FrameDependencies.SpatialId), int32(ddVal.FrameDependencies.TemporalId) } - unwrapped := r.frameWrapAround.Update(ddVal.FrameNumber) + // assume the packet is in-order when stream restarting + unwrapped := r.frameWrapAround.UpdateWithOrderKnown(ddVal.FrameNumber, restart) extFN := unwrapped.ExtendedVal if extFN < r.structureExtFrameNum { @@ -114,9 +144,10 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr r.frameChecker.AddPacket(extSeq, extFN, &ddVal) extDD := &ExtDependencyDescriptor{ - Descriptor: &ddVal, - ExtFrameNum: extFN, - Integrity: r.frameChecker.FrameIntegrity(extFN), + Descriptor: &ddVal, + ExtFrameNum: extFN, + Integrity: r.frameChecker.FrameIntegrity(extFN), + RestartGeneration: r.restartGeneration, } if ddVal.AttachedStructure != nil { @@ -167,6 +198,16 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr return extDD, videoLayer, nil } +func (r *DependencyDescriptorParser) restart() { + r.frameChecker = NewFrameIntegrityChecker(integrityCheckFrame, integrityCheckPkt) + r.structure = nil + r.structureExtFrameNum = 0 + r.activeDecodeTargetsExtSeq = 0 + r.activeDecodeTargetsMask = 0 + r.decodeTargets = r.decodeTargets[:0] + r.restartGeneration++ +} + // ------------------------------------------------------------------------------ type DependencyDescriptorDecodeTarget struct { diff --git a/pkg/sfu/utils/wraparound.go b/pkg/sfu/utils/wraparound.go index c75db3448..78a11999c 100644 --- a/pkg/sfu/utils/wraparound.go +++ b/pkg/sfu/utils/wraparound.go @@ -80,7 +80,7 @@ func (w *WrapAroundUpdateResult[ET]) MarshalLogObject(e zapcore.ObjectEncoder) e return nil } -func (w *WrapAround[T, ET]) Update(val T) (result WrapAroundUpdateResult[ET]) { +func (w *WrapAround[T, ET]) UpdateWithOrderKnown(val T, orderKnown bool) (result WrapAroundUpdateResult[ET]) { if !w.initialized { result.PreExtendedHighest = ET(val) - 1 result.ExtendedVal = ET(val) @@ -92,10 +92,12 @@ func (w *WrapAround[T, ET]) Update(val T) (result WrapAroundUpdateResult[ET]) { return } - gap := val - w.highest - if gap > T(w.fullRange>>1) { - // out-of-order - return w.maybeAdjustStart(val) + if !orderKnown { + gap := val - w.highest + if gap > T(w.fullRange>>1) { + // out-of-order + return w.maybeAdjustStart(val) + } } // in-order @@ -111,6 +113,10 @@ func (w *WrapAround[T, ET]) Update(val T) (result WrapAroundUpdateResult[ET]) { return } +func (w *WrapAround[T, ET]) Update(val T) (result WrapAroundUpdateResult[ET]) { + return w.UpdateWithOrderKnown(val, false) +} + func (w *WrapAround[T, ET]) UndoUpdate(result WrapAroundUpdateResult[ET]) { if !w.initialized || result.PreExtendedHighest >= result.ExtendedVal { return diff --git a/pkg/sfu/videolayerselector/dependencydescriptor.go b/pkg/sfu/videolayerselector/dependencydescriptor.go index 7d93417e2..0cb157093 100644 --- a/pkg/sfu/videolayerselector/dependencydescriptor.go +++ b/pkg/sfu/videolayerselector/dependencydescriptor.go @@ -24,6 +24,11 @@ import ( "github.com/livekit/protocol/logger" ) +const ( + decisionCacheMaxElements = 256 + decisionCacheNackEntries = 80 +) + type DependencyDescriptor struct { *Base @@ -40,12 +45,14 @@ type DependencyDescriptor struct { decodeTargetsLock sync.RWMutex decodeTargets []*DecodeTarget fnWrapper FrameNumberWrapper + + restartGeneration int } func NewDependencyDescriptor(logger logger.Logger) *DependencyDescriptor { return &DependencyDescriptor{ Base: NewBase(logger), - decisions: NewSelectorDecisionCache(256, 80), + decisions: NewSelectorDecisionCache(decisionCacheMaxElements, decisionCacheNackEntries), fnWrapper: FrameNumberWrapper{logger: logger}, } } @@ -75,6 +82,20 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r return } + if ddwdt.RestartGeneration > d.restartGeneration { + d.logger.Debugw("stream restarted", + "packet", ddwdt.RestartGeneration, + "current", d.restartGeneration, + "structureKeyFrame", d.extKeyFrameNum, + "efn", ddwdt.ExtFrameNum, + "lastEfn", d.fnWrapper.LastOrigin(), + ) + d.restart(ddwdt.RestartGeneration) + } else if ddwdt.RestartGeneration < d.restartGeneration { + // must not happen + d.logger.Warnw("packet from old generation", nil, "packet", ddwdt.RestartGeneration, "current", d.restartGeneration) + } + dd := ddwdt.Descriptor extFrameNum := ddwdt.ExtFrameNum @@ -434,3 +455,9 @@ func (d *DependencyDescriptor) CheckSync() (locked bool, layer int32) { return false, layer } + +func (d *DependencyDescriptor) restart(generation int) { + d.restartGeneration = generation + d.invalidateKeyFrame() + d.decisions = NewSelectorDecisionCache(decisionCacheMaxElements, decisionCacheNackEntries) +} diff --git a/pkg/sfu/videolayerselector/framenumberwrapper.go b/pkg/sfu/videolayerselector/framenumberwrapper.go index f0997fc35..43c9c02df 100644 --- a/pkg/sfu/videolayerselector/framenumberwrapper.go +++ b/pkg/sfu/videolayerselector/framenumberwrapper.go @@ -52,3 +52,7 @@ func (f *FrameNumberWrapper) UpdateAndGet(new uint64, updateOffset bool) uint64 f.last = new return new + f.offset } + +func (f *FrameNumberWrapper) LastOrigin() uint64 { + return f.last +}