livekit/pkg/sfu/videolayerselector/framechain.go

// Copyright 2023 LiveKit, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package videolayerselector

import (
	dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor"
	"github.com/livekit/protocol/logger"
)

type FrameChain struct {
	logger         logger.Logger
	decisions      *SelectorDecisionCache
	broken         bool
	chainIdx       int
	active         bool
	updatingActive bool

	expectFrames []uint64
}

func NewFrameChain(decisions *SelectorDecisionCache, chainIdx int, logger logger.Logger) *FrameChain {
	return &FrameChain{
		logger:    logger,
		decisions: decisions,
		broken:    true,
		chainIdx:  chainIdx,
		active:    false,
	}
}

func (fc *FrameChain) OnFrame(extFrameNum uint64, fd *dd.FrameDependencyTemplate) bool {
	if !fc.active {
		return false
	}

	if len(fd.ChainDiffs) <= fc.chainIdx {
		fc.logger.Warnw("invalid frame chain diff", nil, "chanIdx", fc.chainIdx, "frame", extFrameNum, "fd", fd)
		return fc.broken
	}

	// A decodable frame with frame_chain_fdiff equal to 0 indicates that the Chain is intact.
	if fd.ChainDiffs[fc.chainIdx] == 0 {
		if fc.broken {
			fc.broken = false
			// fc.logger.Debugw("frame chain intact", "chanIdx", fc.chainIdx, "frame", extFrameNum)
		}
		fc.expectFrames = fc.expectFrames[:0]
		return true
	}

	if fc.broken {
		return false
	}

	prevFrameInChain := extFrameNum - uint64(fd.ChainDiffs[fc.chainIdx])
	sd, err := fc.decisions.GetDecision(prevFrameInChain)
	if err != nil {
		fc.logger.Debugw("could not get decision", "err", err, "chanIdx", fc.chainIdx, "frame", extFrameNum, "prevFrame", prevFrameInChain)
	}

	var intact bool
	switch sd {
	case selectorDecisionForwarded:
		intact = true

	case selectorDecisionUnknown:
		// If the previous frame is unknown, means it has not arrived but could be recovered by NACK / out-of-order arrival,
		// set up a expected callback here to determine if the chain is broken or intact
		if fc.decisions.ExpectDecision(prevFrameInChain, fc.OnExpectFrameChanged) {
			intact = true
			fc.expectFrames = append(fc.expectFrames, prevFrameInChain)
		}
	}

	if !intact {
		fc.broken = true
		// fc.logger.Debugw("frame chain broken", "chanIdx", fc.chainIdx, "sd", sd, "frame", extFrameNum, "prevFrame", prevFrameInChain)
	}
	return intact
}

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]
			break
		}
	}
}

func (fc *FrameChain) Broken() bool {
	return fc.broken
}

func (fc *FrameChain) BeginUpdateActive() {
	fc.updatingActive = false
}

func (fc *FrameChain) UpdateActive(active bool) {
	fc.updatingActive = fc.updatingActive || active
}

func (fc *FrameChain) EndUpdateActive() {
	active := fc.updatingActive
	fc.updatingActive = false

	if active == fc.active {
		return
	}

	// if the chain transit from inactive to active, reset broken to wait a decodable SWITCH frame
	if !fc.active {
		fc.broken = true
		fc.logger.Debugw("frame chain broken by inactive", "chanIdx", fc.chainIdx)
	}

	fc.active = active
}