add support for av1 svc (#669)

* add support for av1 svc
This commit is contained in:
cnderrauber
2022-05-06 18:24:29 +08:00
committed by GitHub
parent 7f38164ef6
commit 07b93e2e5b
20 changed files with 2107 additions and 243 deletions
+13 -9
View File
@@ -47,15 +47,12 @@ require (
require (
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.1.1 // indirect
github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
github.com/d5/tengo/v2 v2.10.1
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/eapache/channels v1.1.0 // indirect
github.com/eapache/queue v1.1.0 // indirect
github.com/go-logr/zapr v1.2.3 // indirect
github.com/golang/protobuf v1.5.2 // indirect
github.com/google/subcommands v1.2.0 // indirect
github.com/google/uuid v1.3.0 // indirect
github.com/jxskiss/base62 v0.0.0-20191017122030-4f11678b909b // indirect
github.com/lithammer/shortuuid/v3 v3.0.6 // indirect
@@ -72,17 +69,24 @@ require (
github.com/prometheus/client_model v0.2.0 // indirect
github.com/prometheus/common v0.26.0 // indirect
github.com/prometheus/procfs v0.6.0 // indirect
github.com/russross/blackfriday/v2 v2.1.0 // indirect
go.uber.org/multierr v1.6.0 // indirect
golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 // indirect
golang.org/x/mod v0.5.1 // indirect
golang.org/x/net v0.0.0-20220420153159-1850ba15e1be // indirect
golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect
golang.org/x/text v0.3.7 // indirect
golang.org/x/tools v0.1.7 // indirect
golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f // indirect
google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect
google.golang.org/grpc v1.42.0 // indirect
gopkg.in/square/go-jose.v2 v2.5.1 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
)
require (
github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
github.com/d5/tengo/v2 v2.10.1
github.com/google/subcommands v1.2.0 // indirect
github.com/rs/zerolog v1.26.1
github.com/russross/blackfriday/v2 v2.1.0 // indirect
golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 // indirect
golang.org/x/mod v0.5.1 // indirect
golang.org/x/net v0.0.0-20220420153159-1850ba15e1be // indirect
golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect
golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f // indirect
)
+6
View File
@@ -25,6 +25,7 @@ github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XP
github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
github.com/cpuguy83/go-md2man/v2 v2.0.0 h1:EoUDS0afbrsXAZ9YQ9jdu/mZ2sXgT1/2yyNng4PGlyM=
github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
@@ -69,6 +70,7 @@ github.com/go-redis/redis/v8 v8.11.3 h1:GCjoYp8c+yQTJfc0n69iwSiHjvuAdruxl7elnZCx
github.com/go-redis/redis/v8 v8.11.3/go.mod h1:xNJ9xDG09FsIPwh3bWdk+0oDWHbtF9rPN0F/oD9XeKc=
github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
@@ -235,6 +237,9 @@ github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1
github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
github.com/rs/cors v1.8.2 h1:KCooALfAYGs415Cwu5ABvv9n9509fSiG5SQJn/AQo4U=
github.com/rs/cors v1.8.2/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
github.com/rs/xid v1.3.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
github.com/rs/zerolog v1.26.1 h1:/ihwxqH+4z8UxyI70wM1z9yCvkWcfz/a3mj48k/Zngc=
github.com/rs/zerolog v1.26.1/go.mod h1:/wSSJWX7lVrsOwlbyTRSOJvqRlc+WjWlfes+CiJ+tmc=
github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -287,6 +292,7 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
golang.org/x/crypto v0.0.0-20211215165025-cf75a172585e/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8=
golang.org/x/crypto v0.0.0-20220131195533-30dcbda58838/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 h1:kUhD7nTDoI3fVd9G4ORWrbV5NY0liEs/Jg2pv5f+bBA=
golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+6
View File
@@ -11,6 +11,8 @@ import (
"github.com/livekit/protocol/logger"
dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
"github.com/livekit/livekit-server/pkg/config"
logging "github.com/livekit/livekit-server/pkg/logger"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
@@ -147,6 +149,7 @@ func NewWebRTCConfig(conf *config.Config, externalIP string) (*WebRTCConfig, err
sdp.SDESRTPStreamIDURI,
sdp.TransportCCURI,
frameMarking,
dd.ExtensionUrl,
},
},
RTCPFeedback: RTCPFeedbackConfig{
@@ -161,6 +164,9 @@ func NewWebRTCConfig(conf *config.Config, externalIP string) (*WebRTCConfig, err
// subscriber configuration
subscriberConfig := DirectionConfig{
RTPHeaderExtension: RTPHeaderExtensionConfig{
Video: []string{dd.ExtensionUrl},
},
RTCPFeedback: RTCPFeedbackConfig{
Video: []webrtc.RTCPFeedback{
{Type: webrtc.TypeRTCPFBCCM, Parameter: "fir"},
+13 -1
View File
@@ -55,9 +55,13 @@ func ToProtoParticipants(participants []types.LocalParticipant) []*livekit.Parti
}
func ToProtoSessionDescription(sd webrtc.SessionDescription) *livekit.SessionDescription {
sdp := sd.SDP
// enable mixed of one-byte and two-byte rtp extension
// TODO : should move this to pion's option
sdp = strings.Replace(sdp, "m=", "a=extmap-allow-mixed\r\nm=", 1)
return &livekit.SessionDescription{
Type: sd.Type.String(),
Sdp: sd.SDP,
Sdp: sdp,
}
}
@@ -168,3 +172,11 @@ func LoggerWithPCTarget(l logger.Logger, target livekit.SignalTarget) logger.Log
lr = lr.WithValues("transport", target)
return logger.Logger(lr)
}
func LoggerWithCodecMime(l logger.Logger, mime string) logger.Logger {
lr := logr.Logger(l)
if mime != "" {
lr = lr.WithValues("mime", mime)
}
return logger.Logger(lr)
}
+58 -21
View File
@@ -19,6 +19,8 @@ import (
"github.com/livekit/livekit-server/pkg/sfu/twcc"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
)
const (
@@ -31,13 +33,13 @@ type pendingPacket struct {
}
type ExtPacket struct {
Arrival int64
Packet *rtp.Packet
Payload interface{}
KeyFrame bool
RawPacket []byte
SpatialLayer int32
TemporalLayer int32
VideoLayer
Arrival int64
Packet *rtp.Packet
Payload interface{}
KeyFrame bool
RawPacket []byte
DependencyDescriptor *dd.DependencyDescriptor
}
// Buffer contains all packets
@@ -85,6 +87,11 @@ type Buffer struct {
// logger
logger logger.Logger
// depencency descriptor
ddExt uint8
ddParser *DependencyDescriptorParser
maxLayerChangedCB func(int32, int32)
}
// BufferOptions provides configuration options for the buffer
@@ -159,6 +166,21 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
default:
b.codecType = webrtc.RTPCodecType(0)
}
for _, ext := range params.HeaderExtensions {
switch ext.URI {
case dd.ExtensionUrl:
b.ddExt = uint8(ext.ID)
b.ddParser = NewDependencyDescriptorParser(b.ddExt, b.logger, func(spatial, temporal int32) {
if b.maxLayerChangedCB != nil {
b.maxLayerChangedCB(spatial, temporal)
}
})
case sdp.AudioLevelURI:
b.audioLevelExt = uint8(ext.ID)
b.audioLevel = audio.NewAudioLevel(b.audioLevelParams)
}
}
if b.codecType == webrtc.RTPCodecTypeVideo {
for _, fb := range codec.RTCPFeedback {
@@ -180,13 +202,6 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
b.nacker.SetRTT(70) // default till it is updated
}
}
} else if b.codecType == webrtc.RTPCodecTypeAudio {
for _, h := range params.HeaderExtensions {
if h.URI == sdp.AudioLevelURI {
b.audioLevelExt = uint8(h.ID)
b.audioLevel = audio.NewAudioLevel(b.audioLevelParams)
}
}
}
for _, pp := range b.pPackets {
@@ -423,11 +438,13 @@ func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) {
func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTime int64) *ExtPacket {
ep := &ExtPacket{
Packet: rtpPacket,
Arrival: arrivalTime,
RawPacket: rawPacket,
SpatialLayer: -1,
TemporalLayer: -1,
Packet: rtpPacket,
Arrival: arrivalTime,
RawPacket: rawPacket,
VideoLayer: VideoLayer{
Spatial: -1,
Temporal: -1,
},
}
if len(rtpPacket.Payload) == 0 {
@@ -435,7 +452,15 @@ func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTi
return ep
}
ep.TemporalLayer = 0
ep.Temporal = 0
if b.ddParser != nil {
ddVal, videoLayer, err := b.ddParser.Parse(ep.Packet)
if err == nil && ddVal != nil {
ep.DependencyDescriptor = ddVal
ep.VideoLayer = videoLayer
// TODO : notify active decode target change if changed.
}
}
switch b.mime {
case "video/vp8":
vp8Packet := VP8{}
@@ -445,7 +470,13 @@ func (b *Buffer) getExtPacket(rawPacket []byte, rtpPacket *rtp.Packet, arrivalTi
}
ep.Payload = vp8Packet
ep.KeyFrame = vp8Packet.IsKeyFrame
ep.TemporalLayer = int32(vp8Packet.TID)
if ep.DependencyDescriptor == nil {
ep.Temporal = int32(vp8Packet.TID)
} else {
// vp8 with DependencyDescriptor enabled, use the TID from the descriptor
vp8Packet.TID = uint8(ep.Temporal)
ep.Spatial = -1 // vp8 don't have spatial scalability, reset to -1
}
case "video/h264":
ep.KeyFrame = IsH264Keyframe(rtpPacket.Payload)
@@ -625,3 +656,9 @@ func (b *Buffer) GetAudioLevel() (float64, bool) {
return b.audioLevel.GetLevel()
}
// TODO : now we rely on stream tracker for layer change, dependency still
// work for that too. Do we keep it unchange or use both methods?
func (b *Buffer) OnMaxLayerChanged(fn func(int32, int32)) {
b.maxLayerChangedCB = fn
}
@@ -0,0 +1,102 @@
package buffer
import (
"fmt"
"github.com/pion/rtp"
dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
"github.com/livekit/protocol/logger"
)
type DependencyDescriptorParser struct {
structure *dd.FrameDependencyStructure
ddExt uint8
logger logger.Logger
onMaxLayerChanged func(int32, int32)
decodeTargetLayer []VideoLayer
}
func NewDependencyDescriptorParser(ddExt uint8, logger logger.Logger, onMaxLayerChanged func(int32, int32)) *DependencyDescriptorParser {
logger.Infow("creating video stream receiver", "ddExt", ddExt)
return &DependencyDescriptorParser{
ddExt: ddExt,
logger: logger,
onMaxLayerChanged: onMaxLayerChanged,
}
}
func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*dd.DependencyDescriptor, VideoLayer, error) {
var videoLayer VideoLayer
if ddBuf := pkt.GetExtension(r.ddExt); ddBuf != nil {
var ddVal dd.DependencyDescriptor
ext := &dd.DependencyDescriptorExtension{
Descriptor: &ddVal,
Structure: r.structure,
}
_, err := ext.Unmarshal(ddBuf)
if err != nil {
r.logger.Infow("failed to parse generic dependency descriptor", "err", err)
return nil, videoLayer, err
}
if ddVal.FrameDependencies != nil {
videoLayer.Spatial, videoLayer.Temporal = int32(ddVal.FrameDependencies.SpatialId), int32(ddVal.FrameDependencies.TemporalId)
}
if ddVal.AttachedStructure != nil && !ddVal.FirstPacketInFrame {
r.logger.Infow("ignoring non-first packet in frame with attached structure")
return nil, videoLayer, nil
}
if ddVal.AttachedStructure != nil {
var maxSpatial, maxTemporal int32
r.structure = ddVal.AttachedStructure
r.decodeTargetLayer = r.decodeTargetLayer[:0]
for target := 0; target < r.structure.NumDecodeTargets; target++ {
layer := VideoLayer{0, 0}
for _, t := range r.structure.Templates {
if t.DecodeTargetIndications[target] != dd.DecodeTargetNotPresent {
if layer.Spatial < int32(t.SpatialId) {
layer.Spatial = int32(t.SpatialId)
}
if layer.Temporal < int32(t.TemporalId) {
layer.Temporal = int32(t.TemporalId)
}
}
}
if layer.Spatial > maxSpatial {
maxSpatial = layer.Spatial
}
if layer.Temporal > maxTemporal {
maxTemporal = layer.Temporal
}
r.decodeTargetLayer = append(r.decodeTargetLayer, layer)
}
r.logger.Debugw("max layer changed", "maxSpatial", maxSpatial, "maxTemporal", maxTemporal)
go r.onMaxLayerChanged(maxSpatial, maxTemporal)
}
if ddVal.AttachedStructure != nil && ddVal.FirstPacketInFrame {
r.logger.Debugw(fmt.Sprintf("parsed dependency descriptor\n%s", ddVal.String()))
}
if mask := ddVal.ActiveDecodeTargetsBitmask; mask != nil {
var maxSpatial, maxTemporal int32
for dt, layer := range r.decodeTargetLayer {
if *mask&(1<<dt) != uint32(dd.DecodeTargetNotPresent) {
if maxSpatial < layer.Spatial {
maxSpatial = layer.Spatial
}
if maxTemporal < layer.Temporal {
maxTemporal = layer.Temporal
}
}
}
r.logger.Debugw("max layer changed", "maxSpatial", maxSpatial, "maxTemporal", maxTemporal)
r.onMaxLayerChanged(maxSpatial, maxTemporal)
}
return &ddVal, videoLayer, nil
}
return nil, videoLayer, nil
}
+39
View File
@@ -0,0 +1,39 @@
package buffer
import "fmt"
const (
InvalidLayerSpatial = int32(-1)
InvalidLayerTemporal = int32(-1)
DefaultMaxLayerSpatial = int32(2)
DefaultMaxLayerTemporal = int32(3)
)
var (
InvalidLayers = VideoLayer{
Spatial: InvalidLayerSpatial,
Temporal: InvalidLayerTemporal,
}
)
type VideoLayer struct {
Spatial int32
Temporal int32
}
func (v VideoLayer) String() string {
return fmt.Sprintf("VideoLayer{s: %d, t: %d}", v.Spatial, v.Temporal)
}
func (v VideoLayer) GreaterThan(v2 VideoLayer) bool {
return v.Spatial > v2.Spatial || (v.Spatial == v2.Spatial && v.Temporal > v2.Temporal)
}
func (v VideoLayer) SpatialGreaterThanOrEqual(v2 VideoLayer) bool {
return v.Spatial >= v2.Spatial
}
func (v VideoLayer) IsValid() bool {
return v.Spatial != InvalidLayerSpatial && v.Temporal != InvalidLayerTemporal
}
@@ -0,0 +1,124 @@
package dependencydescriptor
import (
"errors"
"io"
)
type BitStreamReader struct {
buf []byte
pos int
remaingBits int
}
func NewBitStreamReader(buf []byte) *BitStreamReader {
return &BitStreamReader{buf: buf, remaingBits: len(buf) * 8}
}
func (b *BitStreamReader) RemaningBits() int {
return b.remaingBits
}
// Reads `bits` from the bitstream. `bits` must be in range [0, 64].
// Returns an unsigned integer in range [0, 2^bits - 1].
// On failure sets `BitstreamReader` into the failure state and returns 0.
func (b *BitStreamReader) ReadBits(bits int) (uint64, error) {
if bits < 0 || bits > 64 {
return 0, errors.New("invalid number of bits, expected 0-64")
}
if b.remaingBits < bits {
b.remaingBits -= bits
return 0, io.EOF
}
remainingBitsInFirstByte := b.remaingBits % 8
b.remaingBits -= bits
if bits < remainingBitsInFirstByte {
// Reading fewer bits than what's left in the current byte, just
// return the portion of this byte that is needed.
offset := (remainingBitsInFirstByte - bits)
return uint64((b.buf[b.pos] >> offset) & ((1 << bits) - 1)), nil
}
var result uint64
if remainingBitsInFirstByte > 0 {
// Read all bits that were left in the current byte and consume that byte.
bits -= remainingBitsInFirstByte
mask := byte((1 << remainingBitsInFirstByte) - 1)
result = uint64(b.buf[b.pos]&mask) << bits
b.pos++
}
// Read as many full bytes as we can.
for bits >= 8 {
bits -= 8
result |= uint64(b.buf[b.pos]) << bits
b.pos++
}
// Whatever is left to read is smaller than a byte, so grab just the needed
// bits and shift them into the lowest bits.
if bits > 0 {
result |= uint64(b.buf[b.pos] >> (8 - bits))
}
return result, nil
}
func (b *BitStreamReader) ReadBool() (bool, error) {
val, err := b.ReadBits(1)
return val != 0, err
}
func (b *BitStreamReader) Ok() bool {
return b.remaingBits >= 0
}
func (b *BitStreamReader) Invalidate() {
b.remaingBits = -1
}
// Reads value in range [0, `num_values` - 1].
// This encoding is similar to ReadBits(val, Ceil(Log2(num_values)),
// but reduces wastage incurred when encoding non-power of two value ranges
// Non symmetric values are encoded as:
// 1) n = bit_width(num_values)
// 2) k = (1 << n) - num_values
// Value v in range [0, k - 1] is encoded in (n-1) bits.
// Value v in range [k, num_values - 1] is encoded as (v+k) in n bits.
// https://aomediacodec.github.io/av1-spec/#nsn
func (b *BitStreamReader) ReadNonSymmetric(numValues uint32) (uint32, error) {
if numValues >= (uint32(1) << 31) {
return 0, errors.New("invalid number of values, expected 0-2^31")
}
width := bitwidth(numValues)
numMinBitsValues := (uint32(1) << width) - numValues
val, err := b.ReadBits(width - 1)
if err != nil {
return 0, err
}
if val < uint64(numMinBitsValues) {
return uint32(val), nil
}
bit, err := b.ReadBits(1)
if err != nil {
return 0, err
}
return uint32((val << 1) + bit - uint64(numMinBitsValues)), nil
}
func (b *BitStreamReader) ReadedBytes() int {
if b.remaingBits%8 > 0 {
return b.pos + 1
}
return b.pos
}
func bitwidth(n uint32) int {
var w int
for n != 0 {
n >>= 1
w++
}
return w
}
@@ -0,0 +1,118 @@
package dependencydescriptor
import (
"errors"
"fmt"
)
type BitStreamWriter struct {
buf []byte
pos int
bitOffset int // bit offset in the current byte
}
func NewBitStreamWriter(buf []byte) *BitStreamWriter {
return &BitStreamWriter{buf: buf}
}
func (w *BitStreamWriter) RemainingBits() int {
return (len(w.buf)-w.pos)*8 - w.bitOffset
}
func (w *BitStreamWriter) WriteBits(val uint64, bitCount int) error {
if bitCount > w.RemainingBits() {
return errors.New("insufficient space")
}
totalBits := bitCount
// push bits to the highest bits of uint64
val <<= (64 - bitCount)
buf := w.buf[w.pos:]
// The first byte is relatively special; the bit offset to write to may put us
// in the middle of the byte, and the total bit count to write may require we
// save the bits at the end of the byte.
remainingBitsInCurrentByte := 8 - w.bitOffset
bitsInFirstByte := bitCount
if bitsInFirstByte > remainingBitsInCurrentByte {
bitsInFirstByte = remainingBitsInCurrentByte
}
buf[0] = w.writePartialByte(uint8(val>>56), bitsInFirstByte, buf[0], w.bitOffset)
if bitCount <= remainingBitsInCurrentByte {
// no bit left to write
return w.consumeBits(totalBits)
}
// write the rest of the bits
val <<= bitsInFirstByte
buf = buf[1:]
bitCount -= bitsInFirstByte
for bitCount >= 8 {
buf[0] = uint8(val >> 56)
buf = buf[1:]
val <<= 8
bitCount -= 8
}
// write the last bits
if bitCount > 0 {
buf[0] = w.writePartialByte(uint8(val>>56), bitCount, buf[0], 0)
}
return w.consumeBits(totalBits)
}
func (w *BitStreamWriter) consumeBits(bitCount int) error {
if bitCount > w.RemainingBits() {
return errors.New("insufficient space")
}
w.pos += (w.bitOffset + bitCount) / 8
w.bitOffset = (w.bitOffset + bitCount) % 8
return nil
}
func (w *BitStreamWriter) writePartialByte(source uint8, sourceBitCount int, target uint8, targetBitOffset int) uint8 {
// if !(targetBitOffset < 8 && sourceBitCount <= (8-targetBitOffset)) {
// return fmt.Errorf("invalid argument, source %d, sourceBitCount %d, target %d, targetBitOffset %d", source, sourceBitCount, target, targetBitOffset)
// }
// generate mask for bits to overwrite, shift source bits to highest bits, then position to target bit offset
mask := uint8(0xff<<(8-sourceBitCount)) >> uint8(targetBitOffset)
// clear target bits and write source bits
return (target & ^mask) | (source >> targetBitOffset)
}
func (w *BitStreamWriter) WriteNonSymmetric(val, numValues uint32) error {
if !(val < numValues && numValues <= 1<<31) {
return fmt.Errorf("invalid argument, val %d, numValues %d", val, numValues)
}
if numValues == 1 {
// When there is only one possible value, it requires zero bits to store it.
// But WriteBits doesn't support writing zero bits.
return nil
}
countBits := bitwidth(numValues)
numMinBitsValues := (uint32(1) << countBits) - numValues
if val < numMinBitsValues {
return w.WriteBits(uint64(val), countBits-1)
} else {
return w.WriteBits(uint64(val+numMinBitsValues), countBits)
}
}
func SizeNonSymmetricBits(val, numValues uint32) int {
countBits := bitwidth(numValues)
numMinBitsValues := (uint32(1) << countBits) - numValues
if val < numMinBitsValues {
return countBits - 1
} else {
return countBits
}
}
@@ -0,0 +1,162 @@
package dependencydescriptor
import (
"fmt"
"math"
"strconv"
)
// DependencyDescriptorExtension is a extension payload format in
// https://aomediacodec.github.io/av1-rtp-spec/#dependency-descriptor-rtp-header-extension
type DependencyDescriptorExtension struct {
Descriptor *DependencyDescriptor
Structure *FrameDependencyStructure
}
func (d *DependencyDescriptor) MarshalSize() (int, error) {
return d.MarshalSizeWithActiveChains(^uint32(0))
}
func (d *DependencyDescriptor) MarshalSizeWithActiveChains(activeChains uint32) (int, error) {
writer, err := NewDependencyDescriptorWriter(nil, d.AttachedStructure, activeChains, d)
if err != nil {
return 0, err
}
return int(math.Ceil(float64(writer.ValueSizeBits()) / 8)), nil
}
func (d *DependencyDescriptorExtension) Marshal() ([]byte, error) {
return d.MarshalWithActiveChains(^uint32(0))
}
func (d *DependencyDescriptorExtension) MarshalWithActiveChains(activeChains uint32) ([]byte, error) {
writer, err := NewDependencyDescriptorWriter(nil, d.Structure, activeChains, d.Descriptor)
if err != nil {
return nil, err
}
buf := make([]byte, int(math.Ceil(float64(writer.ValueSizeBits())/8)))
writer.ResetBuf(buf)
if err = writer.Write(); err != nil {
return nil, err
}
return buf, nil
}
func (d *DependencyDescriptorExtension) Unmarshal(buf []byte) (int, error) {
reader := NewDependencyDescriptorReader(buf, d.Structure, d.Descriptor)
return reader.Parse()
}
const (
MaxSpatialIds = 4
MaxTemporalIds = 8
MaxDecodeTargets = 32
MaxTemplates = 64
AllChainsAreActive = uint32(0)
ExtensionUrl = "https://aomediacodec.github.io/av1-rtp-spec/#dependency-descriptor-rtp-header-extension"
)
type DependencyDescriptor struct {
FirstPacketInFrame bool // = true;
LastPacketInFrame bool // = true;
FrameNumber uint16
FrameDependencies *FrameDependencyTemplate
Resolution *RenderResolution
ActiveDecodeTargetsBitmask *uint32
AttachedStructure *FrameDependencyStructure
}
func formatBitmask(b *uint32) string {
if b == nil {
return "-"
}
return strconv.FormatInt(int64(*b), 2)
}
func (d *DependencyDescriptor) String() string {
return fmt.Sprintf("DependencyDescriptor{FirstPacketInFrame: %v, LastPacketInFrame: %v, FrameNumber: %v, FrameDependencies: %+v, Resolution: %+v, ActiveDecodeTargetsBitmask: %v, AttachedStructure: %v}",
d.FirstPacketInFrame, d.LastPacketInFrame, d.FrameNumber, *d.FrameDependencies, *d.Resolution, formatBitmask(d.ActiveDecodeTargetsBitmask), d.AttachedStructure)
}
// Relationship of a frame to a Decode target.
type DecodeTargetIndication int
const (
DecodeTargetNotPresent DecodeTargetIndication = iota // DecodeTargetInfo symbol '-'
DecodeTargetDiscadable // DecodeTargetInfo symbol 'D'
DecodeTargetSwitch // DecodeTargetInfo symbol 'S'
DecodeTargetRequired // DecodeTargetInfo symbol 'R'
)
func (i DecodeTargetIndication) String() string {
switch i {
case DecodeTargetNotPresent:
return "-"
case DecodeTargetDiscadable:
return "D"
case DecodeTargetSwitch:
return "S"
case DecodeTargetRequired:
return "R"
default:
return "Unknown"
}
}
type FrameDependencyTemplate struct {
SpatialId int
TemporalId int
DecodeTargetIndications []DecodeTargetIndication
FrameDiffs []int
ChainDiffs []int
}
func (t *FrameDependencyTemplate) Clone() *FrameDependencyTemplate {
t2 := &FrameDependencyTemplate{
SpatialId: t.SpatialId,
TemporalId: t.TemporalId,
}
t2.DecodeTargetIndications = make([]DecodeTargetIndication, len(t.DecodeTargetIndications))
copy(t2.DecodeTargetIndications, t.DecodeTargetIndications)
t2.FrameDiffs = make([]int, len(t.FrameDiffs))
copy(t2.FrameDiffs, t.FrameDiffs)
t2.ChainDiffs = make([]int, len(t.ChainDiffs))
copy(t2.ChainDiffs, t.ChainDiffs)
return t2
}
type FrameDependencyStructure struct {
StructureId int
NumDecodeTargets int
NumChains int
// If chains are used (num_chains > 0), maps decode target index into index of
// the chain protecting that target.
DecodeTargetProtectedByChain []int
Resolutions []RenderResolution
Templates []*FrameDependencyTemplate
}
func (f *FrameDependencyStructure) String() string {
str := fmt.Sprintf("FrameDependencyStructure{StructureId: %v, NumDecodeTargets: %v, NumChains: %v, DecodeTargetProtectedByChain: %v, Resolutions: %+v, Templates: [",
f.StructureId, f.NumDecodeTargets, f.NumChains, f.DecodeTargetProtectedByChain, f.Resolutions)
// templates
for _, t := range f.Templates {
str += fmt.Sprintf("%+v, ", t)
}
str += "]}"
return str
}
type RenderResolution struct {
Width int
Height int
}
@@ -0,0 +1,51 @@
package dependencydescriptor
import (
"encoding/hex"
"testing"
)
func TestDependencyDescriptorUnmarshal(t *testing.T) {
// hex bytes from traffic capture
hexes := []string{
"c1017280081485214eafffaaaa863cf0430c10c302afc0aaa0063c00430010c002a000a80006000040001d954926e082b04a0941b820ac1282503157f974000ca864330e222222eca8655304224230eca877530077004200ef008601df010d",
"86017340fc",
"46017340fc",
"c3017540fc",
"88017640fc",
"48017640fc",
"c2017840fc",
//
"c1017280081485214eafffaaaa863cf0430c10c302afc0aaa0063c00430010c002a000a80006000040001d954926e082b04a0941b820ac1282503157f974000ca864330e222222eca8655304224230eca877530077004200ef008601df010d",
"860173",
"460173",
"8b0174",
"0b0174",
"0b0174",
"c30175",
}
var structure *FrameDependencyStructure
for _, h := range hexes {
buf, err := hex.DecodeString(h)
if err != nil {
t.Fatal(err)
}
var ddVal DependencyDescriptor
var d DependencyDescriptorExtension = DependencyDescriptorExtension{
Structure: structure,
Descriptor: &ddVal,
}
if _, err := d.Unmarshal(buf); err != nil {
t.Fatal(err)
}
if ddVal.AttachedStructure != nil {
structure = ddVal.AttachedStructure
}
t.Log(ddVal.String())
}
}
@@ -0,0 +1,418 @@
package dependencydescriptor
import "errors"
type DependencyDescriptorReader struct {
// Output.
descriptor *DependencyDescriptor
// Values that are needed while reading the descriptor, but can be discarded
// when reading is complete.
buffer *BitStreamReader
frameDependencyTemplateId int
activeDecodeTargetsPresentFlag bool
customDtisFlag bool
customFdiffsFlag bool
customChainsFlag bool
structure *FrameDependencyStructure
}
func NewDependencyDescriptorReader(buf []byte, structure *FrameDependencyStructure, descriptor *DependencyDescriptor) *DependencyDescriptorReader {
buffer := NewBitStreamReader(buf)
return &DependencyDescriptorReader{
buffer: buffer,
descriptor: descriptor,
structure: structure,
}
}
func (r *DependencyDescriptorReader) Parse() (int, error) {
r.readMandatoryFields()
if len(r.buffer.buf) > 3 {
err := r.readExtendedFields()
if err != nil {
return 0, err
}
}
if r.descriptor.AttachedStructure != nil {
r.structure = r.descriptor.AttachedStructure
}
if r.structure == nil {
r.buffer.Invalidate()
return 0, errors.New("DependencyDescriptorReader: Structure is nil")
}
if r.activeDecodeTargetsPresentFlag {
bitmask, err := r.buffer.ReadBits(r.structure.NumDecodeTargets)
if err != nil {
return 0, err
}
mask := uint32(bitmask)
r.descriptor.ActiveDecodeTargetsBitmask = &mask
}
err := r.readFrameDependencyDefinition()
if err != nil {
return 0, err
}
return r.buffer.ReadedBytes(), nil
}
func (r *DependencyDescriptorReader) readMandatoryFields() error {
var err error
r.descriptor.FirstPacketInFrame, err = r.buffer.ReadBool()
if err != nil {
return err
}
r.descriptor.LastPacketInFrame, err = r.buffer.ReadBool()
if err != nil {
return err
}
templateID, err := r.buffer.ReadBits(6)
if err != nil {
return err
}
r.frameDependencyTemplateId = int(templateID)
frameNumber, err := r.buffer.ReadBits(16)
if err != nil {
return err
}
r.descriptor.FrameNumber = uint16(frameNumber)
return nil
}
func (r *DependencyDescriptorReader) readExtendedFields() error {
templateDependencyStructurePresentFlag, err := r.buffer.ReadBool()
if err != nil {
return err
}
flag, err := r.buffer.ReadBool()
if err != nil {
return err
}
r.activeDecodeTargetsPresentFlag = flag
flag, err = r.buffer.ReadBool()
if err != nil {
return err
}
r.customDtisFlag = flag
flag, err = r.buffer.ReadBool()
if err != nil {
return err
}
r.customFdiffsFlag = flag
flag, err = r.buffer.ReadBool()
if err != nil {
return err
}
r.customChainsFlag = flag
if templateDependencyStructurePresentFlag {
err = r.readTemplateDependencyStructure()
if err != nil {
return err
}
if r.descriptor.AttachedStructure == nil {
return errors.New("DependencyDescriptorReader: has templateDependencyStructurePresentFlag but AttachedStructure is nil")
}
bitmask := uint32((uint64(1) << r.descriptor.AttachedStructure.NumDecodeTargets) - 1)
r.descriptor.ActiveDecodeTargetsBitmask = &bitmask
}
return nil
}
func (r *DependencyDescriptorReader) readTemplateDependencyStructure() error {
r.descriptor.AttachedStructure = &FrameDependencyStructure{}
structureId, err := r.buffer.ReadBits(6)
if err != nil {
return err
}
r.descriptor.AttachedStructure.StructureId = int(structureId)
numDecodeTargets, err := r.buffer.ReadBits(5)
if err != nil {
return err
}
r.descriptor.AttachedStructure.NumDecodeTargets = int(numDecodeTargets) + 1
if err = r.readTemplateLayers(); err != nil {
return err
}
if err = r.readTemplateDtis(); err != nil {
return err
}
if err = r.readTemplateFdiffs(); err != nil {
return err
}
if err = r.readTemplateChains(); err != nil {
return err
}
flag, err := r.buffer.ReadBool()
if err != nil {
return err
}
if flag {
return r.readResolutions()
}
return nil
}
type nextLayerIdcType int
const (
sameLayer nextLayerIdcType = iota
nextTemporalLayer
nextSpatialLayer
noMoreLayer
invalidLayer
)
func (r *DependencyDescriptorReader) readTemplateLayers() error {
var (
templates []*FrameDependencyTemplate
temporalId, spatialId int
nextLayerIdc nextLayerIdcType
)
for {
if len(templates) == MaxTemplates {
return errors.New("DependencyDescriptorReader: too many templates")
}
var lastTemplate FrameDependencyTemplate
templates = append(templates, &lastTemplate)
lastTemplate.TemporalId = temporalId
lastTemplate.SpatialId = spatialId
idc, err := r.buffer.ReadBits(2)
if err != nil {
return err
}
nextLayerIdc = nextLayerIdcType(idc)
if nextLayerIdc == nextTemporalLayer {
temporalId++
if temporalId >= MaxTemporalIds {
return errors.New("DependencyDescriptorReader: too many temporal layers")
}
} else if nextLayerIdc == nextSpatialLayer {
spatialId++
temporalId = 0
if spatialId >= MaxSpatialIds {
return errors.New("DependencyDescriptorReader: too many spatial layers")
}
}
if !(nextLayerIdc != noMoreLayer && r.buffer.Ok()) {
break
}
}
r.descriptor.AttachedStructure.Templates = templates
return nil
}
func (r *DependencyDescriptorReader) readTemplateDtis() error {
structure := r.descriptor.AttachedStructure
for _, template := range structure.Templates {
if len(template.DecodeTargetIndications) < structure.NumDecodeTargets {
template.DecodeTargetIndications = append(template.DecodeTargetIndications, make([]DecodeTargetIndication, structure.NumDecodeTargets-len(template.DecodeTargetIndications))...)
} else {
template.DecodeTargetIndications = template.DecodeTargetIndications[:structure.NumDecodeTargets]
}
for i := range template.DecodeTargetIndications {
indication, err := r.buffer.ReadBits(2)
if err != nil {
return err
}
template.DecodeTargetIndications[i] = DecodeTargetIndication(indication)
}
}
return nil
}
func (r *DependencyDescriptorReader) readTemplateFdiffs() error {
for _, template := range r.descriptor.AttachedStructure.Templates {
for {
fdiffFollow, err := r.buffer.ReadBool()
if err != nil {
return err
}
if !fdiffFollow {
break
}
fDiffMinusOne, err := r.buffer.ReadBits(4)
if err != nil {
return err
}
template.FrameDiffs = append(template.FrameDiffs, int(fDiffMinusOne+1))
}
}
return nil
}
func (r *DependencyDescriptorReader) readTemplateChains() error {
structure := r.descriptor.AttachedStructure
numChains, err := r.buffer.ReadNonSymmetric(uint32(structure.NumDecodeTargets) + 1)
if err != nil {
return err
}
structure.NumChains = int(numChains)
if structure.NumChains == 0 {
return nil
}
for i := 0; i < structure.NumDecodeTargets; i++ {
protectedByChain, err := r.buffer.ReadNonSymmetric(uint32(structure.NumChains))
if err != nil {
return err
}
structure.DecodeTargetProtectedByChain = append(structure.DecodeTargetProtectedByChain, int(protectedByChain))
}
for _, frameTemplate := range structure.Templates {
for chainId := 0; chainId < structure.NumChains; chainId++ {
chainDiff, err := r.buffer.ReadBits(4)
if err != nil {
return err
}
frameTemplate.ChainDiffs = append(frameTemplate.ChainDiffs, int(chainDiff))
}
}
return nil
}
func (r *DependencyDescriptorReader) readResolutions() error {
structure := r.descriptor.AttachedStructure
// The way templates are bitpacked, they are always ordered by spatial_id.
spatialLayers := structure.Templates[len(structure.Templates)-1].SpatialId + 1
for sid := 0; sid < spatialLayers; sid++ {
widthMinus1, err := r.buffer.ReadBits(16)
if err != nil {
return err
}
heightMinus1, err := r.buffer.ReadBits(16)
if err != nil {
return err
}
structure.Resolutions = append(structure.Resolutions, RenderResolution{
Width: int(widthMinus1 + 1),
Height: int(heightMinus1 + 1),
})
}
return nil
}
func (r *DependencyDescriptorReader) readFrameDependencyDefinition() error {
templateIndex := (r.frameDependencyTemplateId + MaxTemplates - r.structure.StructureId) % MaxTemplates
if templateIndex >= len(r.structure.Templates) {
r.buffer.Invalidate()
return errors.New("DependencyDescriptorReader: invalid template index")
}
// Copy all the fields from the matching template
r.descriptor.FrameDependencies = r.structure.Templates[templateIndex].Clone()
if r.customDtisFlag {
err := r.readFrameDtis()
if err != nil {
return err
}
}
if r.customFdiffsFlag {
err := r.readFrameFdiffs()
if err != nil {
return err
}
}
if r.customChainsFlag {
err := r.readFrameChains()
if err != nil {
return err
}
}
if len(r.structure.Resolutions) == 0 {
r.descriptor.Resolution = nil
} else {
// Format guarantees that if there were resolutions in the last structure,
// then each spatial layer got one.
if r.descriptor.FrameDependencies.SpatialId >= len(r.structure.Resolutions) {
r.buffer.Invalidate()
return errors.New("DependencyDescriptorReader: invalid spatial layer, should be less than the number of resolutions")
}
res := r.structure.Resolutions[r.descriptor.FrameDependencies.SpatialId]
r.descriptor.Resolution = &res
}
return nil
}
func (r *DependencyDescriptorReader) readFrameDtis() error {
if len(r.descriptor.FrameDependencies.DecodeTargetIndications) != r.structure.NumDecodeTargets {
return errors.New("DependencyDescriptorReader: decode target indications length mismatch with structure num decode targets")
}
for i := range r.descriptor.FrameDependencies.DecodeTargetIndications {
indication, err := r.buffer.ReadBits(2)
if err != nil {
return err
}
r.descriptor.FrameDependencies.DecodeTargetIndications[i] = DecodeTargetIndication(indication)
}
return nil
}
func (r *DependencyDescriptorReader) readFrameFdiffs() error {
r.descriptor.FrameDependencies.FrameDiffs = r.descriptor.FrameDependencies.FrameDiffs[:0]
for {
nexFdiffSize, err := r.buffer.ReadBits(2)
if err != nil {
return err
}
if nexFdiffSize == 0 {
break
}
fDiffMinusOne, err := r.buffer.ReadBits(int(nexFdiffSize * 4))
if err != nil {
return err
}
r.descriptor.FrameDependencies.FrameDiffs = append(r.descriptor.FrameDependencies.FrameDiffs, int(fDiffMinusOne+1))
}
return nil
}
func (r *DependencyDescriptorReader) readFrameChains() error {
if len(r.descriptor.FrameDependencies.ChainDiffs) != r.structure.NumChains {
return errors.New("DependencyDescriptorReader: chain diffs length mismatch with structure num chains")
}
for i := range r.descriptor.FrameDependencies.ChainDiffs {
chainDiff, err := r.buffer.ReadBits(8)
if err != nil {
return err
}
r.descriptor.FrameDependencies.ChainDiffs[i] = int(chainDiff)
}
return nil
}
@@ -0,0 +1,501 @@
package dependencydescriptor
import (
"fmt"
"reflect"
)
type TemplateMatch struct {
TemplateIdx int
NeedCustomDtis bool
NeedCustomFdiffs bool
NeedCustomChains bool
// Size in bits to store frame-specific details, i.e.
// excluding mandatory fields and template dependency structure.
ExtraSizeBits int
}
type DependencyDescriptorWriter struct {
descriptor *DependencyDescriptor
structure *FrameDependencyStructure
activeChains uint32
writer *BitStreamWriter
bestTemplate TemplateMatch
}
func NewDependencyDescriptorWriter(buf []byte, structure *FrameDependencyStructure, activeChains uint32, descriptor *DependencyDescriptor) (*DependencyDescriptorWriter, error) {
writer := NewBitStreamWriter(buf)
w := &DependencyDescriptorWriter{
descriptor: descriptor,
structure: structure,
activeChains: activeChains,
writer: writer,
}
return w, w.findBestTemplate()
}
func (w *DependencyDescriptorWriter) ResetBuf(buf []byte) {
w.writer = NewBitStreamWriter(buf)
}
func (w *DependencyDescriptorWriter) Write() error {
if err := w.findBestTemplate(); err != nil {
return err
}
if err := w.writeMandatoryFields(); err != nil {
return err
}
if w.hasExtenedFields() {
if err := w.writeExtendedFields(); err != nil {
return err
}
if err := w.writeFrameDependencyDefinition(); err != nil {
return err
}
}
remaingBits := w.writer.RemainingBits()
// Zero remaining memory to avoid leaving it uninitialized.
if remaingBits%64 != 0 {
if err := w.writeBits(0, remaingBits%64); err != nil {
return err
}
}
for i := 0; i < remaingBits/64; i++ {
if err := w.writeBits(0, 64); err != nil {
return err
}
}
return nil
}
func (w *DependencyDescriptorWriter) findBestTemplate() error {
// Find templates with same spatial and temporal layer of frame dependency.
var (
firstSameLayer *FrameDependencyTemplate
firstSameLayerIdx, lastSameLayerIdx int
)
for i, t := range w.structure.Templates {
if w.descriptor.FrameDependencies.SpatialId == t.SpatialId &&
w.descriptor.FrameDependencies.TemporalId == t.TemporalId {
firstSameLayer = t
firstSameLayerIdx = i
break
}
}
if firstSameLayer == nil {
return fmt.Errorf("no template found for spatial layer %d and temporal layer %d", w.descriptor.FrameDependencies.SpatialId, w.descriptor.FrameDependencies.TemporalId)
}
for i, t := range w.structure.Templates[firstSameLayerIdx:] {
if w.descriptor.FrameDependencies.SpatialId != t.SpatialId ||
w.descriptor.FrameDependencies.TemporalId != t.TemporalId {
lastSameLayerIdx = i + firstSameLayerIdx
}
}
// Search if there any better template that have small extra size.
w.bestTemplate = w.caculateMatch(firstSameLayerIdx, firstSameLayer)
for i := firstSameLayerIdx + 1; i <= lastSameLayerIdx; i++ {
t := w.structure.Templates[i]
match := w.caculateMatch(i, t)
if match.ExtraSizeBits < w.bestTemplate.ExtraSizeBits {
w.bestTemplate = match
}
}
return nil
}
// TODO: uncomment this when go 1.18 enabled
// func sliceEqual[T comparable](a, b []T) bool {
// if len(a) != len(b) {
// return false
// }
// for i, v := range a {
// if v != b[i] {
// return false
// }
// }
// return true
// }
func (w *DependencyDescriptorWriter) caculateMatch(idx int, template *FrameDependencyTemplate) TemplateMatch {
var result TemplateMatch
result.TemplateIdx = idx
result.NeedCustomFdiffs = w.descriptor.FrameDependencies.FrameDiffs != nil && !reflect.DeepEqual(w.descriptor.FrameDependencies.FrameDiffs, template.FrameDiffs)
result.NeedCustomDtis = w.descriptor.FrameDependencies.DecodeTargetIndications != nil && !reflect.DeepEqual(w.descriptor.FrameDependencies.DecodeTargetIndications, template.DecodeTargetIndications)
for i := 0; i < w.structure.NumChains; i++ {
if w.activeChains&(1<<i) != 0 && w.descriptor.FrameDependencies.ChainDiffs[i] != template.ChainDiffs[i] {
result.NeedCustomChains = true
break
}
}
if result.NeedCustomFdiffs {
result.ExtraSizeBits = 2 * (1 + len(w.descriptor.FrameDependencies.FrameDiffs))
for _, fdiff := range w.descriptor.FrameDependencies.FrameDiffs {
if fdiff <= (1 << 4) {
result.ExtraSizeBits += 4
} else if fdiff <= (1 << 8) {
result.ExtraSizeBits += 8
} else {
result.ExtraSizeBits += 12
}
}
}
if result.NeedCustomDtis {
result.ExtraSizeBits += 2 * len(w.descriptor.FrameDependencies.DecodeTargetIndications)
}
if result.NeedCustomChains {
result.ExtraSizeBits += 8 * w.structure.NumChains
}
return result
}
func (w *DependencyDescriptorWriter) writeMandatoryFields() error {
if err := w.writeBool(w.descriptor.FirstPacketInFrame); err != nil {
return err
}
if err := w.writeBool(w.descriptor.LastPacketInFrame); err != nil {
return err
}
templateId := (w.bestTemplate.TemplateIdx + w.structure.StructureId) % MaxTemplates
if err := w.writeBits(uint64(templateId), 6); err != nil {
return err
}
return w.writeBits(uint64(w.descriptor.FrameNumber), 16)
}
func (w *DependencyDescriptorWriter) writeBool(val bool) error {
v := uint64(0)
if val {
v = 1
}
return w.writer.WriteBits(v, 1)
}
func (w *DependencyDescriptorWriter) writeBits(val uint64, bitCount int) error {
if err := w.writer.WriteBits(uint64(val), bitCount); err != nil {
return err
}
return nil
}
func (w *DependencyDescriptorWriter) hasExtenedFields() bool {
return w.bestTemplate.ExtraSizeBits > 0 || w.descriptor.AttachedStructure != nil || w.descriptor.ActiveDecodeTargetsBitmask != nil
}
func (w *DependencyDescriptorWriter) writeExtendedFields() error {
// template_dependency_structure_present_flag
if err := w.writeBool(w.descriptor.AttachedStructure != nil); err != nil {
return err
}
// active_decode_targets_present_flag
activeDecodeTargetsPresentFlag := w.shouldWriteActiveDecodeTargetsBitmask()
if err := w.writeBool(activeDecodeTargetsPresentFlag); err != nil {
return err
}
// need_custom_dtis
if err := w.writeBool(w.bestTemplate.NeedCustomDtis); err != nil {
return err
}
// need_custom_fdiffs
if err := w.writeBool(w.bestTemplate.NeedCustomFdiffs); err != nil {
return err
}
// need_custom_chains
if err := w.writeBool(w.bestTemplate.NeedCustomChains); err != nil {
return err
}
// template_dependency_structure
if w.descriptor.AttachedStructure != nil {
if err := w.writeTemplateDependencyStructure(); err != nil {
return err
}
}
// active_decode_targets_bitmask
if activeDecodeTargetsPresentFlag {
if err := w.writeBits(uint64(*w.descriptor.ActiveDecodeTargetsBitmask), w.structure.NumDecodeTargets); err != nil {
return err
}
}
return nil
}
func (w *DependencyDescriptorWriter) writeTemplateDependencyStructure() error {
if !(w.structure.StructureId >= 0 && w.structure.StructureId < MaxTemplates &&
w.structure.NumDecodeTargets > 0 && w.structure.NumDecodeTargets <= MaxDecodeTargets) {
return fmt.Errorf("invalid arguments, structureId: %d, numDecodeTargets: %d", w.structure.StructureId, w.structure.NumDecodeTargets)
}
if err := w.writeBits(uint64(w.structure.StructureId), 6); err != nil {
return err
}
if err := w.writeBits(uint64(w.structure.NumDecodeTargets-1), 5); err != nil {
return err
}
if err := w.writeTemplateLayers(); err != nil {
return err
}
if err := w.writeTemplateDtis(); err != nil {
return err
}
if err := w.writeTemplateFdiffs(); err != nil {
return err
}
if err := w.writeTemplateChains(); err != nil {
return err
}
hasResolutions := len(w.structure.Resolutions) > 0
if err := w.writeBool(hasResolutions); err != nil {
return err
}
return w.writeResolutions()
}
func (w *DependencyDescriptorWriter) writeTemplateLayers() error {
if !(len(w.structure.Templates) > 0 && len(w.structure.Templates) <= MaxTemplates &&
w.structure.Templates[0].SpatialId == 0 && w.structure.Templates[0].TemporalId == 0) {
return fmt.Errorf("invalid templates, len %d, templates[0]: spatialId %d, temporalId %d", len(w.structure.Templates), w.structure.Templates[0].SpatialId, w.structure.Templates[0].TemporalId)
}
for i := 1; i < len(w.structure.Templates); i++ {
nextLayerIdc := getNextLayerIdc(w.structure.Templates[i-1], w.structure.Templates[i])
if nextLayerIdc >= 3 {
return fmt.Errorf("invalid next_layer_idc %d", nextLayerIdc)
}
if err := w.writeBits(uint64(nextLayerIdc), 2); err != nil {
return err
}
}
return w.writeBits(uint64(noMoreLayer), 2)
}
func getNextLayerIdc(prevTemplate, nextTemplate *FrameDependencyTemplate) nextLayerIdcType {
if nextTemplate.SpatialId == prevTemplate.SpatialId && nextTemplate.TemporalId == prevTemplate.TemporalId {
return sameLayer
} else if nextTemplate.SpatialId == prevTemplate.SpatialId && nextTemplate.TemporalId == prevTemplate.TemporalId+1 {
return nextTemporalLayer
} else if nextTemplate.SpatialId == prevTemplate.SpatialId+1 && nextTemplate.TemporalId == 0 {
return nextSpatialLayer
}
return invalidLayer
}
func (w *DependencyDescriptorWriter) writeTemplateDtis() error {
for _, t := range w.structure.Templates {
for _, dti := range t.DecodeTargetIndications {
if err := w.writeBits(uint64(dti), 2); err != nil {
return err
}
}
}
return nil
}
func (w *DependencyDescriptorWriter) writeTemplateFdiffs() error {
for _, t := range w.structure.Templates {
for _, fdiff := range t.FrameDiffs {
if err := w.writeBits(uint64(1<<4)|uint64(fdiff-1), 1+4); err != nil {
return err
}
}
// no more fdiffs for this template
if err := w.writeBits(uint64(0), 1); err != nil {
return err
}
}
return nil
}
func (w *DependencyDescriptorWriter) writeTemplateChains() error {
if err := w.writeNonSymmetric(uint32(w.structure.NumChains), uint32(w.structure.NumDecodeTargets+1)); err != nil {
return err
}
if w.structure.NumChains == 0 {
return nil
}
for _, protectedBy := range w.structure.DecodeTargetProtectedByChain {
if err := w.writeNonSymmetric(uint32(protectedBy), uint32(w.structure.NumChains)); err != nil {
return err
}
}
for _, t := range w.structure.Templates {
for _, chainDiff := range t.ChainDiffs {
if err := w.writeBits(uint64(chainDiff), 4); err != nil {
return err
}
}
}
return nil
}
func (w *DependencyDescriptorWriter) writeNonSymmetric(value, numValues uint32) error {
return w.writer.WriteNonSymmetric(value, numValues)
}
func (w *DependencyDescriptorWriter) writeResolutions() error {
for _, res := range w.structure.Resolutions {
if err := w.writeBits(uint64(res.Width)-1, 16); err != nil {
return err
}
if err := w.writeBits(uint64(res.Height)-1, 16); err != nil {
return err
}
}
return nil
}
func (w *DependencyDescriptorWriter) writeFrameDependencyDefinition() error {
if w.bestTemplate.NeedCustomDtis {
if err := w.writeFrameDtis(); err != nil {
return err
}
}
if w.bestTemplate.NeedCustomFdiffs {
if err := w.writeFrameFdiffs(); err != nil {
return err
}
}
if w.bestTemplate.NeedCustomChains {
if err := w.writeFrameChains(); err != nil {
return err
}
}
return nil
}
func (w *DependencyDescriptorWriter) writeFrameDtis() error {
for _, dti := range w.descriptor.FrameDependencies.DecodeTargetIndications {
if err := w.writeBits(uint64(dti), 2); err != nil {
return err
}
}
return nil
}
func (w *DependencyDescriptorWriter) writeFrameFdiffs() error {
for _, fdiff := range w.descriptor.FrameDependencies.FrameDiffs {
if fdiff <= (1 << 4) {
if err := w.writeBits(uint64(1<<4)|uint64(fdiff-1), 2+4); err != nil {
return err
}
} else if fdiff <= (1 << 8) {
if err := w.writeBits(uint64(2<<8)|uint64(fdiff-1), 2+8); err != nil {
return err
}
} else { // fdiff <= (1<<12)
if err := w.writeBits(uint64(3<<12)|uint64(fdiff-1), 2+12); err != nil {
return err
}
}
}
// no more fdiffs
return w.writeBits(uint64(0), 2)
}
func (w *DependencyDescriptorWriter) writeFrameChains() error {
for i := 0; i < w.structure.NumChains; i++ {
chainDiff := 0
if w.activeChains&(1<<i) != 0 {
chainDiff = w.descriptor.FrameDependencies.ChainDiffs[i]
}
if err := w.writeBits(uint64(chainDiff), 8); err != nil {
return err
}
}
return nil
}
const mandatoryFieldSize = 1 + 1 + 6 + 16
func (w *DependencyDescriptorWriter) ValueSizeBits() int {
valueSizeBits := mandatoryFieldSize + w.bestTemplate.ExtraSizeBits
if w.hasExtenedFields() {
valueSizeBits += 5
if w.descriptor.AttachedStructure != nil {
valueSizeBits += w.structureSizeBits()
}
if w.shouldWriteActiveDecodeTargetsBitmask() {
valueSizeBits += w.structure.NumDecodeTargets
}
}
return valueSizeBits
}
func (w *DependencyDescriptorWriter) shouldWriteActiveDecodeTargetsBitmask() bool {
if w.descriptor.ActiveDecodeTargetsBitmask == nil {
return false
}
allDecodeTargetsBitmask := (uint64(1) << w.structure.NumDecodeTargets) - 1
if w.descriptor.AttachedStructure != nil && uint64(*w.descriptor.ActiveDecodeTargetsBitmask) == allDecodeTargetsBitmask {
return false
}
return true
}
func (w *DependencyDescriptorWriter) structureSizeBits() int {
// template_id offset (6 bits) and number of decode targets (5 bits)
bits := 11
// template layers
bits += 2 * len(w.structure.Templates)
// dtis
bits += 2 * len(w.structure.Templates) * w.structure.NumDecodeTargets
// fdiffs. each templates uses 1 + 5 * sizeof(fdiff) bits.
bits += len(w.structure.Templates)
for _, t := range w.structure.Templates {
bits += 5 * len(t.FrameDiffs)
}
bits += SizeNonSymmetricBits(uint32(w.structure.NumChains), uint32(w.structure.NumDecodeTargets+1))
if w.structure.NumChains > 0 {
for _, protectedBy := range w.structure.DecodeTargetProtectedByChain {
bits += SizeNonSymmetricBits(uint32(protectedBy), uint32(w.structure.NumChains))
}
bits += 4 * len(w.structure.Templates) * w.structure.NumChains
}
// resolutions
bits += 1 + 32*len(w.structure.Resolutions)
return bits
}
+53 -18
View File
@@ -21,6 +21,7 @@ import (
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
)
// TrackSender defines an interface send media to remote peer
@@ -92,6 +93,8 @@ type DownTrack struct {
codec webrtc.RTPCodecCapability
rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter
absSendTimeID int
dependencyDescriptorID int
receiver TrackReceiver
transceiver *webrtc.RTPTransceiver
writeStream webrtc.TrackLocalWriter
@@ -229,7 +232,7 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
d.handleRTCP(pkt)
})
}
if strings.HasPrefix(d.codec.MimeType, "video/") {
if strings.HasPrefix(codec.MimeType, "video/") {
d.sequencer = newSequencer(d.maxTrack, d.logger)
}
@@ -267,6 +270,14 @@ func (d *DownTrack) PeerID() livekit.ParticipantID { return d.peerID }
// Sets RTP header extensions for this track
func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter) {
d.rtpHeaderExtensions = rtpHeaderExtensions
for _, ext := range rtpHeaderExtensions {
switch ext.URI {
case sdp.ABSSendTimeURI:
d.absSendTimeID = ext.ID
case dd.ExtensionUrl:
d.dependencyDescriptorID = ext.ID
}
}
}
// Kind controls if this TrackLocal is audio or video
@@ -303,6 +314,7 @@ func (d *DownTrack) maybeStartKeyFrameRequester() {
//
d.stopKeyFrameRequester()
// TODO : for svc, don't need pli/lrr when layer comes down
locked, layer := d.forwarder.CheckSync()
if !locked {
go d.keyFrameRequester(d.keyFrameRequestGeneration.Load(), layer)
@@ -381,7 +393,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
}
}
hdr, err := d.getTranslatedRTPHeader(extPkt, tp.rtp)
hdr, err := d.getTranslatedRTPHeader(extPkt, tp)
if err != nil {
d.pktsDropped.Inc()
return err
@@ -398,7 +410,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
d.onMaxLayerChanged(d, layer)
}
if extPkt.KeyFrame {
if extPkt.KeyFrame || tp.switchingToTargetLayer {
d.isNACKThrottled.Store(false)
d.rtpStats.UpdateKeyFrame(1)
@@ -407,7 +419,10 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
d.stopKeyFrameRequester()
}
d.logger.Debugw("forwarding key frame", "layer", layer)
// too much log for switching target layer, only log key frame
if !tp.switchingToTargetLayer {
d.logger.Debugw("forwarding key frame", "layer", layer)
}
}
d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano())
@@ -529,7 +544,7 @@ func (d *DownTrack) Mute(muted bool) {
// client might need to be notified to start layers
// before locking can happen in the forwarder.
//
notifyLayer = maxLayers.spatial
notifyLayer = maxLayers.Spatial
}
d.onMaxLayerChanged(d, notifyLayer)
}
@@ -606,7 +621,7 @@ func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) {
// a. is higher than previous max -> client may need to start higher layer before forwarder can lock
// b. is lower than previous max -> client can stop higher layer(s)
//
d.onMaxLayerChanged(d, maxLayers.spatial)
d.onMaxLayerChanged(d, maxLayers.Spatial)
}
if d.onSubscribedLayersChanged != nil {
@@ -924,8 +939,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
if pliOnce {
targetLayers := d.forwarder.TargetLayers()
if targetLayers != InvalidLayers {
d.logger.Debugw("sending PLI RTCP", "layer", targetLayers.spatial)
d.receiver.SendPLI(targetLayers.spatial)
d.logger.Debugw("sending PLI RTCP", "layer", targetLayers.Spatial)
d.receiver.SendPLI(targetLayers.Spatial)
d.isNACKThrottled.Store(true)
d.rtpStats.UpdatePliTime()
pliOnce = false
@@ -1125,26 +1140,30 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
d.rtpStats.UpdateNackProcessed(nackAcks, nackMisses, numRepeatedNACKs)
}
type extensionData struct {
id uint8
payload []byte
}
// writes RTP header extensions of track
func (d *DownTrack) writeRTPHeaderExtensions(hdr *rtp.Header) error {
func (d *DownTrack) writeRTPHeaderExtensions(hdr *rtp.Header, extraExtensions ...extensionData) error {
// clear out extensions that may have been in the forwarded header
hdr.Extension = false
hdr.ExtensionProfile = 0
hdr.Extensions = []rtp.Extension{}
for _, ext := range d.rtpHeaderExtensions {
if ext.URI != sdp.ABSSendTimeURI {
// supporting only abs-send-time
continue
}
for _, ext := range extraExtensions {
hdr.SetExtension(uint8(ext.id), ext.payload)
}
if d.absSendTimeID != 0 {
sendTime := rtp.NewAbsSendTimeExtension(time.Now())
b, err := sendTime.Marshal()
if err != nil {
return err
}
err = hdr.SetExtension(uint8(ext.ID), b)
err = hdr.SetExtension(uint8(d.absSendTimeID), b)
if err != nil {
return err
}
@@ -1153,14 +1172,30 @@ func (d *DownTrack) writeRTPHeaderExtensions(hdr *rtp.Header) error {
return nil
}
func (d *DownTrack) getTranslatedRTPHeader(extPkt *buffer.ExtPacket, tpRTP *TranslationParamsRTP) (*rtp.Header, error) {
func (d *DownTrack) getTranslatedRTPHeader(extPkt *buffer.ExtPacket, tp *TranslationParams) (*rtp.Header, error) {
tpRTP := tp.rtp
hdr := extPkt.Packet.Header
hdr.PayloadType = d.payloadType
hdr.Timestamp = tpRTP.timestamp
hdr.SequenceNumber = tpRTP.sequenceNumber
hdr.SSRC = d.ssrc
if tp.marker {
hdr.Marker = tp.marker
}
err := d.writeRTPHeaderExtensions(&hdr)
var extension []extensionData
if d.dependencyDescriptorID != 0 && tp.ddExtension != nil {
bytes, err := tp.ddExtension.Marshal()
if err != nil {
d.logger.Infow("marshalling dependency descriptor extension err", "err", err)
} else {
extension = append(extension, extensionData{
id: uint8(d.dependencyDescriptorID),
payload: bytes,
})
}
}
err := d.writeRTPHeaderExtensions(&hdr, extension...)
if err != nil {
return nil, err
}
@@ -1212,7 +1247,7 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
"MimeType": d.codec.MimeType,
"Bound": d.bound.Load(),
"Muted": d.forwarder.IsMuted(),
"CurrentSpatialLayer": d.forwarder.CurrentLayers().spatial,
"CurrentSpatialLayer": d.forwarder.CurrentLayers().Spatial,
"Stats": stats,
}
}
+122 -108
View File
@@ -11,6 +11,7 @@ import (
"github.com/livekit/protocol/logger"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
)
//
@@ -135,46 +136,28 @@ type TranslationParams struct {
isSwitchingToMaxLayer bool
rtp *TranslationParamsRTP
vp8 *TranslationParamsVP8
ddExtension *dd.DependencyDescriptorExtension
marker bool
// indicates this frame has 'Switch' decode indication for target layer
// TODO : in theory, we need check frame chain is not broken for the target
// but we don't have frame queue now, so just use decode target indication
switchingToTargetLayer bool
}
// -------------------------------------------------------------------
type VideoLayers struct {
spatial int32
temporal int32
}
func (v VideoLayers) String() string {
return fmt.Sprintf("VideoLayers{s: %d, t: %d}", v.spatial, v.temporal)
}
func (v VideoLayers) GreaterThan(v2 VideoLayers) bool {
return v.spatial > v2.spatial || (v.spatial == v2.spatial && v.temporal > v2.temporal)
}
func (v VideoLayers) SpatialGreaterThanOrEqual(v2 VideoLayers) bool {
return v.spatial >= v2.spatial
}
func (v VideoLayers) IsValid() bool {
return v.spatial != InvalidLayerSpatial && v.temporal != InvalidLayerTemporal
}
// -------------------------------------------------------------------
type VideoLayers = buffer.VideoLayer
const (
InvalidLayerSpatial = int32(-1)
InvalidLayerTemporal = int32(-1)
InvalidLayerSpatial = buffer.InvalidLayerSpatial
InvalidLayerTemporal = buffer.InvalidLayerTemporal
DefaultMaxLayerSpatial = int32(2)
DefaultMaxLayerTemporal = int32(3)
DefaultMaxLayerSpatial = buffer.DefaultMaxLayerSpatial
DefaultMaxLayerTemporal = buffer.DefaultMaxLayerTemporal
)
var (
InvalidLayers = VideoLayers{
spatial: InvalidLayerSpatial,
temporal: InvalidLayerTemporal,
}
InvalidLayers = buffer.InvalidLayers
)
// -------------------------------------------------------------------
@@ -205,6 +188,8 @@ type Forwarder struct {
vp8Munger *VP8Munger
isTemporalSupported bool
ddLayerSelector *DDVideoLayerSelector
}
func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType, logger logger.Logger) *Forwarder {
@@ -227,8 +212,14 @@ func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType, log
f.vp8Munger = NewVP8Munger(logger)
}
// TODO : we only enable dd layer selector for av1 now, at future we can
// enable it for vp9 too
if strings.ToLower(codec.MimeType) == "video/av1" {
f.ddLayerSelector = NewDDVideoLayerSelector(f.logger)
}
if f.kind == webrtc.RTPCodecTypeVideo {
f.maxLayers = VideoLayers{spatial: InvalidLayerSpatial, temporal: DefaultMaxLayerTemporal}
f.maxLayers = VideoLayers{Spatial: InvalidLayerSpatial, Temporal: DefaultMaxLayerTemporal}
} else {
f.maxLayers = InvalidLayers
}
@@ -265,11 +256,11 @@ func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, VideoLayers, V
f.lock.Lock()
defer f.lock.Unlock()
if f.kind == webrtc.RTPCodecTypeAudio || spatialLayer == f.maxLayers.spatial {
if f.kind == webrtc.RTPCodecTypeAudio || spatialLayer == f.maxLayers.Spatial {
return false, f.maxLayers, f.currentLayers
}
f.maxLayers.spatial = spatialLayer
f.maxLayers.Spatial = spatialLayer
return true, f.maxLayers, f.currentLayers
}
@@ -278,11 +269,11 @@ func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, VideoLayers,
f.lock.Lock()
defer f.lock.Unlock()
if f.kind == webrtc.RTPCodecTypeAudio || temporalLayer == f.maxLayers.temporal {
if f.kind == webrtc.RTPCodecTypeAudio || temporalLayer == f.maxLayers.Temporal {
return false, f.maxLayers, f.currentLayers
}
f.maxLayers.temporal = temporalLayer
f.maxLayers.Temporal = temporalLayer
return true, f.maxLayers, f.currentLayers
}
@@ -320,7 +311,7 @@ func (f *Forwarder) GetForwardingStatus() ForwardingStatus {
return ForwardingStatusOff
}
if f.targetLayers.spatial < f.maxLayers.spatial && f.targetLayers.spatial < f.availableLayers[len(f.availableLayers)-1] {
if f.targetLayers.Spatial < f.maxLayers.Spatial && f.targetLayers.Spatial < f.availableLayers[len(f.availableLayers)-1] {
return ForwardingStatusPartial
}
@@ -335,8 +326,8 @@ func (f *Forwarder) UpTrackLayersChange(availableLayers []int32) {
}
func (f *Forwarder) getOptimalBandwidthNeeded(brs Bitrates) int64 {
for i := f.maxLayers.spatial; i >= 0; i-- {
for j := f.maxLayers.temporal; j >= 0; j-- {
for i := f.maxLayers.Spatial; i >= 0; i-- {
for j := f.maxLayers.Temporal; j >= 0; j-- {
if brs[i][j] == 0 {
continue
}
@@ -352,12 +343,12 @@ func (f *Forwarder) bitrateAvailable(brs Bitrates, availableLayers []int32) bool
neededLayers := 0
var bitrateAvailableLayers []int32
for _, layer := range f.availableLayers {
if layer > f.maxLayers.spatial {
if layer > f.maxLayers.Spatial {
continue
}
neededLayers++
for t := f.maxLayers.temporal; t >= 0; t-- {
for t := f.maxLayers.Temporal; t >= 0; t-- {
if brs[layer][t] != 0 {
bitrateAvailableLayers = append(bitrateAvailableLayers, layer)
break
@@ -374,13 +365,13 @@ func (f *Forwarder) getDistanceToDesired(brs Bitrates, targetLayers VideoLayers)
}
distance := int32(0)
for s := f.maxLayers.spatial; s >= 0; s-- {
for s := f.maxLayers.Spatial; s >= 0; s-- {
found := false
for t := f.maxLayers.temporal; t >= 0; t-- {
for t := f.maxLayers.Temporal; t >= 0; t-- {
if brs[s][t] == 0 {
continue
}
if s == targetLayers.spatial && t == targetLayers.temporal {
if s == targetLayers.Spatial && t == targetLayers.Temporal {
found = true
break
}
@@ -411,7 +402,7 @@ func (f *Forwarder) BandwidthRequested(brs Bitrates) int64 {
return 0
}
return brs[f.targetLayers.spatial][f.targetLayers.temporal]
return brs[f.targetLayers.Spatial][f.targetLayers.Temporal]
}
func (f *Forwarder) DistanceToDesired() int32 {
@@ -448,23 +439,23 @@ func (f *Forwarder) AllocateOptimal(brs Bitrates) VideoAllocation {
// Resume with the highest layer available <= max subscribed layer
// If already resumed, move allocation to the highest available layer <= max subscribed layer
//
targetLayers.spatial = int32(math.Min(float64(f.maxLayers.spatial), float64(f.availableLayers[len(f.availableLayers)-1])))
targetLayers.temporal = int32(math.Max(0, float64(f.maxLayers.temporal)))
targetLayers.Spatial = int32(math.Min(float64(f.maxLayers.Spatial), float64(f.availableLayers[len(f.availableLayers)-1])))
targetLayers.Temporal = int32(math.Max(0, float64(f.maxLayers.Temporal)))
if f.targetLayers == InvalidLayers && targetLayers.IsValid() {
change = VideoStreamingChangeResuming
}
default:
// allocate best layer available
for s := f.maxLayers.spatial; s >= 0; s-- {
for t := f.maxLayers.temporal; t >= 0; t-- {
for s := f.maxLayers.Spatial; s >= 0; s-- {
for t := f.maxLayers.Temporal; t >= 0; t-- {
if brs[s][t] == 0 {
continue
}
targetLayers = VideoLayers{
spatial: s,
temporal: t,
Spatial: s,
Temporal: t,
}
bandwidthRequested = brs[s][t]
@@ -495,7 +486,7 @@ func (f *Forwarder) AllocateOptimal(brs Bitrates) VideoAllocation {
targetLayers: targetLayers,
distanceToDesired: f.getDistanceToDesired(brs, targetLayers),
}
f.targetLayers = f.lastAllocation.targetLayers
f.setTargetLayers(f.lastAllocation.targetLayers)
if f.targetLayers == InvalidLayers {
f.resyncLocked()
}
@@ -524,14 +515,14 @@ func (f *Forwarder) ProvisionalAllocate(availableChannelCapacity int64, layers V
return 0
}
requiredBitrate := f.provisional.bitrates[layers.spatial][layers.temporal]
requiredBitrate := f.provisional.bitrates[layers.Spatial][layers.Temporal]
if requiredBitrate == 0 {
return 0
}
alreadyAllocatedBitrate := int64(0)
if f.provisional.layers != InvalidLayers {
alreadyAllocatedBitrate = f.provisional.bitrates[f.provisional.layers.spatial][f.provisional.layers.temporal]
alreadyAllocatedBitrate = f.provisional.bitrates[f.provisional.layers.Spatial][f.provisional.layers.Temporal]
}
if requiredBitrate <= (availableChannelCapacity + alreadyAllocatedBitrate) {
@@ -586,10 +577,10 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition() VideoTransitio
// what is the highest that is available
maximalLayers := InvalidLayers
maximalBandwidthRequired := int64(0)
for s := f.provisional.maxLayers.spatial; s >= 0; s-- {
for t := f.provisional.maxLayers.temporal; t >= 0; t-- {
for s := f.provisional.maxLayers.Spatial; s >= 0; s-- {
for t := f.provisional.maxLayers.Temporal; t >= 0; t-- {
if f.provisional.bitrates[s][t] != 0 {
maximalLayers = VideoLayers{spatial: s, temporal: t}
maximalLayers = VideoLayers{Spatial: s, Temporal: t}
maximalBandwidthRequired = f.provisional.bitrates[s][t]
break
}
@@ -601,7 +592,7 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition() VideoTransitio
}
if maximalLayers != InvalidLayers {
if !f.targetLayers.GreaterThan(maximalLayers) && (f.provisional.bitrates[f.targetLayers.spatial][f.targetLayers.temporal] != 0) {
if !f.targetLayers.GreaterThan(maximalLayers) && (f.provisional.bitrates[f.targetLayers.Spatial][f.targetLayers.Temporal] != 0) {
// currently streaming and wanting an upgrade, just preserve current target in the cooperative scheme of things
f.provisional.layers = f.targetLayers
return VideoTransition{
@@ -628,10 +619,10 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition() VideoTransitio
// but the cooperative scheme knocks things back to minimal
minimalLayers := InvalidLayers
bandwidthRequired := int64(0)
for s := int32(0); s <= f.provisional.maxLayers.spatial; s++ {
for t := int32(0); t <= f.provisional.maxLayers.temporal; t++ {
for s := int32(0); s <= f.provisional.maxLayers.Spatial; s++ {
for t := int32(0); t <= f.provisional.maxLayers.Temporal; t++ {
if f.provisional.bitrates[s][t] != 0 {
minimalLayers = VideoLayers{spatial: s, temporal: t}
minimalLayers = VideoLayers{Spatial: s, Temporal: t}
bandwidthRequired = f.provisional.bitrates[s][t]
break
}
@@ -643,7 +634,7 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition() VideoTransitio
}
targetLayers := f.targetLayers
if targetLayers == InvalidLayers || targetLayers.GreaterThan(minimalLayers) || (f.provisional.bitrates[targetLayers.spatial][targetLayers.temporal] == 0) {
if targetLayers == InvalidLayers || targetLayers.GreaterThan(minimalLayers) || (f.provisional.bitrates[targetLayers.Spatial][targetLayers.Temporal] == 0) {
targetLayers = minimalLayers
}
@@ -684,8 +675,8 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti
}
maxReachableLayerTemporal := InvalidLayerTemporal
for t := f.provisional.maxLayers.temporal; t >= 0; t-- {
for s := f.provisional.maxLayers.spatial; s >= 0; s-- {
for t := f.provisional.maxLayers.Temporal; t >= 0; t-- {
for s := f.provisional.maxLayers.Spatial; s >= 0; s-- {
if f.provisional.bitrates[s][t] != 0 {
maxReachableLayerTemporal = t
break
@@ -712,20 +703,20 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti
bestLayers := InvalidLayers
bestBandwidthDelta := int64(0)
bestValue := float32(0)
for s := int32(0); s <= f.targetLayers.spatial; s++ {
for t := int32(0); t <= f.targetLayers.temporal; t++ {
if s == f.targetLayers.spatial && t == f.targetLayers.temporal {
for s := int32(0); s <= f.targetLayers.Spatial; s++ {
for t := int32(0); t <= f.targetLayers.Temporal; t++ {
if s == f.targetLayers.Spatial && t == f.targetLayers.Temporal {
break
}
bandwidthDelta := int64(math.Max(float64(0), float64(f.lastAllocation.bandwidthRequested-f.provisional.bitrates[s][t])))
transitionCost := int32(0)
if f.targetLayers.spatial != s {
if f.targetLayers.Spatial != s {
transitionCost = TransitionCostSpatial
}
qualityCost := (maxReachableLayerTemporal+1)*(f.targetLayers.spatial-s) + (f.targetLayers.temporal - t)
qualityCost := (maxReachableLayerTemporal+1)*(f.targetLayers.Spatial-s) + (f.targetLayers.Temporal - t)
value := float32(0)
if (transitionCost + qualityCost) != 0 {
@@ -734,7 +725,7 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti
if value > bestValue || (value == bestValue && bandwidthDelta > bestBandwidthDelta) {
bestValue = value
bestBandwidthDelta = bandwidthDelta
bestLayers = VideoLayers{spatial: s, temporal: t}
bestLayers = VideoLayers{Spatial: s, Temporal: t}
}
}
}
@@ -768,7 +759,7 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation {
change = VideoStreamingChangePausing
}
default:
bandwidthRequested = f.provisional.bitrates[f.provisional.layers.spatial][f.provisional.layers.temporal]
bandwidthRequested = f.provisional.bitrates[f.provisional.layers.Spatial][f.provisional.layers.Temporal]
if bandwidthRequested == f.getOptimalBandwidthNeeded(f.provisional.bitrates) {
state = VideoAllocationStateOptimal
} else {
@@ -790,7 +781,7 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation {
targetLayers: f.provisional.layers,
distanceToDesired: f.getDistanceToDesired(f.provisional.bitrates, f.provisional.layers),
}
f.targetLayers = f.lastAllocation.targetLayers
f.setTargetLayers(f.lastAllocation.targetLayers)
if f.targetLayers == InvalidLayers {
f.resyncLocked()
}
@@ -823,13 +814,13 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra
alreadyAllocated := int64(0)
if f.targetLayers != InvalidLayers {
alreadyAllocated = brs[f.targetLayers.spatial][f.targetLayers.temporal]
alreadyAllocated = brs[f.targetLayers.Spatial][f.targetLayers.Temporal]
}
// try moving temporal layer up in currently streaming spatial layer
if f.targetLayers != InvalidLayers {
for t := f.targetLayers.temporal + 1; t <= f.maxLayers.temporal; t++ {
bandwidthRequested := brs[f.targetLayers.spatial][t]
for t := f.targetLayers.Temporal + 1; t <= f.maxLayers.Temporal; t++ {
bandwidthRequested := brs[f.targetLayers.Spatial][t]
if bandwidthRequested == 0 {
continue
}
@@ -845,7 +836,7 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra
state = VideoAllocationStateDeficient
}
targetLayers := VideoLayers{spatial: f.targetLayers.spatial, temporal: t}
targetLayers := VideoLayers{Spatial: f.targetLayers.Spatial, Temporal: t}
f.lastAllocation = VideoAllocation{
state: state,
change: VideoStreamingChangeNone,
@@ -856,14 +847,14 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra
targetLayers: targetLayers,
distanceToDesired: f.getDistanceToDesired(brs, targetLayers),
}
f.targetLayers = f.lastAllocation.targetLayers
f.setTargetLayers(f.lastAllocation.targetLayers)
return f.lastAllocation, true
}
}
// try moving spatial layer up if temporal layer move up is not available
for s := f.targetLayers.spatial + 1; s <= f.maxLayers.spatial; s++ {
for t := int32(0); t <= f.maxLayers.temporal; t++ {
for s := f.targetLayers.Spatial + 1; s <= f.maxLayers.Spatial; s++ {
for t := int32(0); t <= f.maxLayers.Temporal; t++ {
bandwidthRequested := brs[s][t]
if bandwidthRequested == 0 {
continue
@@ -885,7 +876,7 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra
change = VideoStreamingChangeResuming
}
targetLayers := VideoLayers{spatial: s, temporal: t}
targetLayers := VideoLayers{Spatial: s, Temporal: t}
f.lastAllocation = VideoAllocation{
state: state,
change: change,
@@ -896,7 +887,7 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra
targetLayers: targetLayers,
distanceToDesired: f.getDistanceToDesired(brs, targetLayers),
}
f.targetLayers = f.lastAllocation.targetLayers
f.setTargetLayers(f.lastAllocation.targetLayers)
return f.lastAllocation, true
}
}
@@ -925,20 +916,20 @@ func (f *Forwarder) GetNextHigherTransition(brs Bitrates) (VideoTransition, bool
alreadyAllocated := int64(0)
if f.targetLayers != InvalidLayers {
alreadyAllocated = brs[f.targetLayers.spatial][f.targetLayers.temporal]
alreadyAllocated = brs[f.targetLayers.Spatial][f.targetLayers.Temporal]
}
// try moving temporal layer up in currently streaming spatial layer
if f.targetLayers != InvalidLayers {
for t := f.targetLayers.temporal + 1; t <= f.maxLayers.temporal; t++ {
bandwidthRequested := brs[f.targetLayers.spatial][t]
for t := f.targetLayers.Temporal + 1; t <= f.maxLayers.Temporal; t++ {
bandwidthRequested := brs[f.targetLayers.Spatial][t]
if bandwidthRequested == 0 {
continue
}
transition := VideoTransition{
from: f.targetLayers,
to: VideoLayers{spatial: f.targetLayers.spatial, temporal: t},
to: VideoLayers{Spatial: f.targetLayers.Spatial, Temporal: t},
bandwidthDelta: bandwidthRequested - alreadyAllocated,
}
@@ -947,8 +938,8 @@ func (f *Forwarder) GetNextHigherTransition(brs Bitrates) (VideoTransition, bool
}
// try moving spatial layer up if temporal layer move up is not available
for s := f.targetLayers.spatial + 1; s <= f.maxLayers.spatial; s++ {
for t := int32(0); t <= f.maxLayers.temporal; t++ {
for s := f.targetLayers.Spatial + 1; s <= f.maxLayers.Spatial; s++ {
for t := int32(0); t <= f.maxLayers.Temporal; t++ {
bandwidthRequested := brs[s][t]
if bandwidthRequested == 0 {
continue
@@ -956,7 +947,7 @@ func (f *Forwarder) GetNextHigherTransition(brs Bitrates) (VideoTransition, bool
transition := VideoTransition{
from: f.targetLayers,
to: VideoLayers{spatial: s, temporal: t},
to: VideoLayers{Spatial: s, Temporal: t},
bandwidthDelta: bandwidthRequested - alreadyAllocated,
}
@@ -999,7 +990,7 @@ func (f *Forwarder) Pause(brs Bitrates) VideoAllocation {
targetLayers: InvalidLayers,
distanceToDesired: f.getDistanceToDesired(brs, InvalidLayers),
}
f.targetLayers = f.lastAllocation.targetLayers
f.setTargetLayers(f.lastAllocation.targetLayers)
if f.targetLayers == InvalidLayers {
f.resyncLocked()
}
@@ -1007,6 +998,13 @@ func (f *Forwarder) Pause(brs Bitrates) VideoAllocation {
return f.lastAllocation
}
func (f *Forwarder) setTargetLayers(targetLayers VideoLayers) {
f.targetLayers = targetLayers
if f.ddLayerSelector != nil {
f.ddLayerSelector.SelectLayer(targetLayers)
}
}
func (f *Forwarder) Resync() {
f.lock.Lock()
defer f.lock.Unlock()
@@ -1023,8 +1021,8 @@ func (f *Forwarder) CheckSync() (locked bool, layer int32) {
f.lock.RLock()
defer f.lock.RUnlock()
layer = f.targetLayers.spatial
locked = f.targetLayers.spatial == f.currentLayers.spatial
layer = f.targetLayers.Spatial
locked = f.targetLayers.Spatial == f.currentLayers.Spatial
return
}
@@ -1050,7 +1048,7 @@ func (f *Forwarder) FilterRTX(nacks []uint16) (filtered []uint16, disallowedLaye
//
for layer := int32(0); layer < DefaultMaxLayerSpatial+1; layer++ {
if f.lastAllocation.state == VideoAllocationStateDeficient &&
(f.targetLayers.spatial < f.currentLayers.spatial || layer > f.currentLayers.spatial) {
(f.targetLayers.Spatial < f.currentLayers.Spatial || layer > f.currentLayers.Spatial) {
disallowedLayers[layer] = true
}
}
@@ -1079,7 +1077,7 @@ func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32)
}
// should be called with lock held
func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket) (*TranslationParams, error) {
func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, tp *TranslationParams) (*TranslationParams, error) {
if f.lastSSRC != extPkt.Packet.SSRC {
if !f.started {
f.started = true
@@ -1121,7 +1119,9 @@ func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket) (*Trans
f.lTSCalc = extPkt.Arrival
tp := &TranslationParams{}
if tp == nil {
tp = &TranslationParams{}
}
tpRTP, err := f.rtpMunger.UpdateAndGetSnTs(extPkt)
if err != nil {
tp.shouldDrop = true
@@ -1142,7 +1142,7 @@ func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket) (*Trans
// should be called with lock held
func (f *Forwarder) getTranslationParamsAudio(extPkt *buffer.ExtPacket) (*TranslationParams, error) {
return f.getTranslationParamsCommon(extPkt)
return f.getTranslationParamsCommon(extPkt, nil)
}
// should be called with lock held
@@ -1155,28 +1155,42 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in
return tp, nil
}
if f.targetLayers.spatial != f.currentLayers.spatial {
if f.targetLayers.spatial == layer {
if extPkt.KeyFrame {
if f.ddLayerSelector != nil {
if selected := f.ddLayerSelector.Select(extPkt, tp); !selected {
tp.shouldDrop = true
f.rtpMunger.PacketDropped(extPkt)
return tp, nil
}
}
if f.targetLayers.Spatial != f.currentLayers.Spatial {
if f.targetLayers.Spatial == layer {
if extPkt.KeyFrame || tp.switchingToTargetLayer {
// lock to target layer
f.logger.Debugw("locking to target layer", "current", f.currentLayers, "target", f.targetLayers)
f.currentLayers.spatial = f.targetLayers.spatial
f.currentLayers.Spatial = f.targetLayers.Spatial
if !f.isTemporalSupported {
f.currentLayers.temporal = f.targetLayers.temporal
f.currentLayers.Temporal = f.targetLayers.Temporal
}
if f.currentLayers.spatial == f.maxLayers.spatial {
// TODO : we switch to target layer immediately now since we assume all frame chain is integrity
// if we have frame chain check, should switch only if target chain is not broken and decodable
// if f.ddLayerSelector != nil {
// f.ddLayerSelector.SelectLayer(f.currentLayers)
// }
if f.currentLayers.Spatial == f.maxLayers.Spatial {
tp.isSwitchingToMaxLayer = true
}
}
}
}
if f.currentLayers.spatial != layer {
// if we have layer selector, let it decide whether to drop or not
if f.ddLayerSelector == nil && f.currentLayers.Spatial != layer {
tp.shouldDrop = true
return tp, nil
}
if FlagPauseOnDowngrade && f.targetLayers.spatial < f.currentLayers.spatial && f.lastAllocation.state == VideoAllocationStateDeficient {
if FlagPauseOnDowngrade && f.targetLayers.Spatial < f.currentLayers.Spatial && f.lastAllocation.state == VideoAllocationStateDeficient {
//
// If target layer is lower than both the current and
// maximum subscribed layer, it is due to bandwidth
@@ -1200,22 +1214,22 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in
return tp, nil
}
tp, err := f.getTranslationParamsCommon(extPkt)
_, err := f.getTranslationParamsCommon(extPkt, tp)
if tp.shouldDrop || f.vp8Munger == nil || len(extPkt.Packet.Payload) == 0 {
return tp, err
}
// catch up temporal layer if necessary
if f.currentLayers.temporal != f.targetLayers.temporal {
if f.currentLayers.Temporal != f.targetLayers.Temporal {
incomingVP8, ok := extPkt.Payload.(buffer.VP8)
if ok {
if incomingVP8.TIDPresent == 1 && incomingVP8.TID <= uint8(f.targetLayers.temporal) {
f.currentLayers.temporal = f.targetLayers.temporal
if incomingVP8.TIDPresent == 1 && incomingVP8.TID <= uint8(f.targetLayers.Temporal) {
f.currentLayers.Temporal = f.targetLayers.Temporal
}
}
}
tpVP8, err := f.vp8Munger.UpdateAndGet(extPkt, tp.rtp.snOrdering, f.currentLayers.temporal)
tpVP8, err := f.vp8Munger.UpdateAndGet(extPkt, tp.rtp.snOrdering, f.currentLayers.Temporal)
if err != nil {
tp.rtp = nil
tp.shouldDrop = true
+70 -70
View File
@@ -62,15 +62,15 @@ func TestForwarderLayersVideo(t *testing.T) {
f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo)
maxLayers := f.MaxLayers()
expectedLayers := VideoLayers{spatial: InvalidLayerSpatial, temporal: DefaultMaxLayerTemporal}
expectedLayers := VideoLayers{Spatial: InvalidLayerSpatial, Temporal: DefaultMaxLayerTemporal}
require.Equal(t, expectedLayers, maxLayers)
require.Equal(t, InvalidLayers, f.CurrentLayers())
require.Equal(t, InvalidLayers, f.TargetLayers())
expectedLayers = VideoLayers{
spatial: DefaultMaxLayerSpatial,
temporal: DefaultMaxLayerTemporal,
Spatial: DefaultMaxLayerSpatial,
Temporal: DefaultMaxLayerTemporal,
}
changed, maxLayers, currentLayers := f.SetMaxSpatialLayer(DefaultMaxLayerSpatial)
require.True(t, changed)
@@ -80,34 +80,34 @@ func TestForwarderLayersVideo(t *testing.T) {
changed, maxLayers, currentLayers = f.SetMaxSpatialLayer(DefaultMaxLayerSpatial - 1)
require.True(t, changed)
expectedLayers = VideoLayers{
spatial: DefaultMaxLayerSpatial - 1,
temporal: DefaultMaxLayerTemporal,
Spatial: DefaultMaxLayerSpatial - 1,
Temporal: DefaultMaxLayerTemporal,
}
require.Equal(t, expectedLayers, maxLayers)
require.Equal(t, expectedLayers, f.MaxLayers())
require.Equal(t, InvalidLayers, currentLayers)
f.currentLayers = VideoLayers{spatial: 0, temporal: 1}
f.currentLayers = VideoLayers{Spatial: 0, Temporal: 1}
changed, maxLayers, currentLayers = f.SetMaxSpatialLayer(DefaultMaxLayerSpatial - 1)
require.False(t, changed)
require.Equal(t, expectedLayers, maxLayers)
require.Equal(t, expectedLayers, f.MaxLayers())
require.Equal(t, VideoLayers{spatial: 0, temporal: 1}, currentLayers)
require.Equal(t, VideoLayers{Spatial: 0, Temporal: 1}, currentLayers)
changed, maxLayers, currentLayers = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal)
require.False(t, changed)
require.Equal(t, expectedLayers, maxLayers)
require.Equal(t, VideoLayers{spatial: 0, temporal: 1}, currentLayers)
require.Equal(t, VideoLayers{Spatial: 0, Temporal: 1}, currentLayers)
changed, maxLayers, currentLayers = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal - 1)
require.True(t, changed)
expectedLayers = VideoLayers{
spatial: DefaultMaxLayerSpatial - 1,
temporal: DefaultMaxLayerTemporal - 1,
Spatial: DefaultMaxLayerSpatial - 1,
Temporal: DefaultMaxLayerTemporal - 1,
}
require.Equal(t, expectedLayers, maxLayers)
require.Equal(t, expectedLayers, f.MaxLayers())
require.Equal(t, VideoLayers{spatial: 0, temporal: 1}, currentLayers)
require.Equal(t, VideoLayers{Spatial: 0, Temporal: 1}, currentLayers)
}
func TestForwarderGetForwardingStatus(t *testing.T) {
@@ -129,11 +129,11 @@ func TestForwarderGetForwardingStatus(t *testing.T) {
// when target is the max, should be optimal
f.Mute(false)
f.targetLayers.spatial = DefaultMaxLayerSpatial
f.targetLayers.Spatial = DefaultMaxLayerSpatial
require.Equal(t, ForwardingStatusOptimal, f.GetForwardingStatus())
// when target is less than max subscribed and max available, should be partial
f.targetLayers.spatial = DefaultMaxLayerSpatial - 1
f.targetLayers.Spatial = DefaultMaxLayerSpatial - 1
require.Equal(t, ForwardingStatusPartial, f.GetForwardingStatus())
// when available layers are lower than max subscribed, optimal as long as target is at max available
@@ -212,8 +212,8 @@ func TestForwarderAllocate(t *testing.T) {
disable(f)
f.UpTrackLayersChange([]int32{0})
expectedTargetLayers := VideoLayers{
spatial: 0,
temporal: DefaultMaxLayerTemporal,
Spatial: 0,
Temporal: DefaultMaxLayerTemporal,
}
expectedResult = VideoAllocation{
state: VideoAllocationStateAwaitingMeasurement,
@@ -233,8 +233,8 @@ func TestForwarderAllocate(t *testing.T) {
// allocate using bitrates, allocation should choose optimal
expectedTargetLayers = VideoLayers{
spatial: 2,
temporal: 1,
Spatial: 2,
Temporal: 1,
}
expectedResult = VideoAllocation{
state: VideoAllocationStateOptimal,
@@ -268,26 +268,26 @@ func TestForwarderProvisionalAllocate(t *testing.T) {
f.availableLayers = availableLayers
f.ProvisionalAllocatePrepare(bitrates)
usedBitrate := f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}, true)
usedBitrate := f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 0, Temporal: 0}, true)
require.Equal(t, bitrates[0][0], usedBitrate)
usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 2, temporal: 3}, true)
usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 2, Temporal: 3}, true)
require.Equal(t, bitrates[2][3]-bitrates[0][0], usedBitrate)
usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 3}, true)
usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 0, Temporal: 3}, true)
require.Equal(t, bitrates[0][3]-bitrates[2][3], usedBitrate)
usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 1, temporal: 2}, true)
usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 1, Temporal: 2}, true)
require.Equal(t, bitrates[1][2]-bitrates[0][3], usedBitrate)
// available not enough to reach (2, 2), allocating at (2, 2) should not succeed
usedBitrate = f.ProvisionalAllocate(bitrates[2][2]-bitrates[1][2]-1, VideoLayers{spatial: 2, temporal: 2}, true)
usedBitrate = f.ProvisionalAllocate(bitrates[2][2]-bitrates[1][2]-1, VideoLayers{Spatial: 2, Temporal: 2}, true)
require.Equal(t, int64(0), usedBitrate)
// committing should set target to (1, 2)
expectedTargetLayers := VideoLayers{
spatial: 1,
temporal: 2,
Spatial: 1,
Temporal: 2,
}
expectedResult := VideoAllocation{
state: VideoAllocationStateDeficient,
@@ -307,13 +307,13 @@ func TestForwarderProvisionalAllocate(t *testing.T) {
// when nothing fits and pausing disallowed, should allocate (0, 0)
f.targetLayers = InvalidLayers
f.ProvisionalAllocatePrepare(bitrates)
usedBitrate = f.ProvisionalAllocate(0, VideoLayers{spatial: 0, temporal: 0}, false)
usedBitrate = f.ProvisionalAllocate(0, VideoLayers{Spatial: 0, Temporal: 0}, false)
require.Equal(t, int64(1), usedBitrate)
// committing should set target to (0, 0)
expectedTargetLayers = VideoLayers{
spatial: 0,
temporal: 0,
Spatial: 0,
Temporal: 0,
}
expectedResult = VideoAllocation{
state: VideoAllocationStateDeficient,
@@ -343,10 +343,10 @@ func TestForwarderProvisionalAllocateMute(t *testing.T) {
f.Mute(true)
f.ProvisionalAllocatePrepare(bitrates)
usedBitrate := f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}, true)
usedBitrate := f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 0, Temporal: 0}, true)
require.Equal(t, int64(0), usedBitrate)
usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 1, temporal: 2}, true)
usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 1, Temporal: 2}, true)
require.Equal(t, int64(0), usedBitrate)
// committing should set target to InvalidLayers as track is muted
@@ -384,14 +384,14 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) {
// from scratch (InvalidLayers) should give back layer (0, 0)
expectedTransition := VideoTransition{
from: InvalidLayers,
to: VideoLayers{spatial: 0, temporal: 0},
to: VideoLayers{Spatial: 0, Temporal: 0},
bandwidthDelta: 1,
}
transition := f.ProvisionalAllocateGetCooperativeTransition()
require.Equal(t, expectedTransition, transition)
// committing should set target to (0, 0)
expectedLayers := VideoLayers{spatial: 0, temporal: 0}
expectedLayers := VideoLayers{Spatial: 0, Temporal: 0}
expectedResult := VideoAllocation{
state: VideoAllocationStateDeficient,
change: VideoStreamingChangeResuming,
@@ -408,7 +408,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) {
require.Equal(t, expectedLayers, f.TargetLayers())
// a higher target that is already streaming, just maintain it
targetLayers := VideoLayers{spatial: 2, temporal: 1}
targetLayers := VideoLayers{Spatial: 2, Temporal: 1}
f.targetLayers = targetLayers
f.lastAllocation.bandwidthRequested = 10
expectedTransition = VideoTransition{
@@ -420,7 +420,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) {
require.Equal(t, expectedTransition, transition)
// committing should set target to (2, 1)
expectedLayers = VideoLayers{spatial: 2, temporal: 1}
expectedLayers = VideoLayers{Spatial: 2, Temporal: 1}
expectedResult = VideoAllocation{
state: VideoAllocationStateOptimal,
change: VideoStreamingChangeNone,
@@ -437,11 +437,11 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) {
require.Equal(t, expectedLayers, f.TargetLayers())
// from a target that has become unavailable, should switch to lower available layer
targetLayers = VideoLayers{spatial: 2, temporal: 2}
targetLayers = VideoLayers{Spatial: 2, Temporal: 2}
f.targetLayers = targetLayers
expectedTransition = VideoTransition{
from: targetLayers,
to: VideoLayers{spatial: 2, temporal: 1},
to: VideoLayers{Spatial: 2, Temporal: 1},
bandwidthDelta: 0,
}
transition = f.ProvisionalAllocateGetCooperativeTransition()
@@ -455,7 +455,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) {
// mute should send target to InvalidLayers
expectedTransition = VideoTransition{
from: VideoLayers{spatial: 2, temporal: 1},
from: VideoLayers{Spatial: 2, Temporal: 1},
to: InvalidLayers,
bandwidthDelta: -10,
}
@@ -476,11 +476,11 @@ func TestForwarderProvisionalAllocateGetBestWeightedTransition(t *testing.T) {
f.ProvisionalAllocatePrepare(bitrates)
f.targetLayers = VideoLayers{spatial: 2, temporal: 2}
f.targetLayers = VideoLayers{Spatial: 2, Temporal: 2}
f.lastAllocation.bandwidthRequested = bitrates[2][2]
expectedTransition := VideoTransition{
from: f.targetLayers,
to: VideoLayers{spatial: 2, temporal: 0},
to: VideoLayers{Spatial: 2, Temporal: 0},
bandwidthDelta: 2,
}
transition := f.ProvisionalAllocateGetBestWeightedTransition()
@@ -515,7 +515,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) {
// if layers have not caught up, should not allocate next layer
f.lastAllocation.state = VideoAllocationStateDeficient
f.targetLayers.spatial = 0
f.targetLayers.Spatial = 0
expectedResult := VideoAllocation{
state: VideoAllocationStateDeficient,
change: VideoStreamingChangeNone,
@@ -530,14 +530,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) {
require.Equal(t, expectedResult, result)
require.Equal(t, expectedResult, f.lastAllocation)
require.False(t, boosted)
f.currentLayers.spatial = 0
f.currentLayers.Spatial = 0
f.targetLayers.temporal = 0
f.targetLayers.Temporal = 0
result, boosted = f.AllocateNextHigher(ChannelCapacityInfinity, bitrates)
require.Equal(t, expectedResult, result)
require.Equal(t, expectedResult, f.lastAllocation)
require.False(t, boosted)
f.currentLayers.temporal = 0
f.currentLayers.Temporal = 0
f.lastAllocation.bandwidthRequested = bitrates[0][0]
@@ -559,8 +559,8 @@ func TestForwarderAllocateNextHigher(t *testing.T) {
// move from (0, 0) -> (0, 1), i.e. a higher temporal layer is available in the same spatial layer
expectedTargetLayers := VideoLayers{
spatial: 0,
temporal: 1,
Spatial: 0,
Temporal: 1,
}
expectedResult = VideoAllocation{
state: VideoAllocationStateDeficient,
@@ -579,10 +579,10 @@ func TestForwarderAllocateNextHigher(t *testing.T) {
require.True(t, boosted)
// move from (0, 1) -> (1, 0), i.e. a higher spatial layer is available
f.currentLayers.temporal = 1
f.currentLayers.Temporal = 1
expectedTargetLayers = VideoLayers{
spatial: 1,
temporal: 0,
Spatial: 1,
Temporal: 0,
}
expectedResult = VideoAllocation{
state: VideoAllocationStateDeficient,
@@ -601,11 +601,11 @@ func TestForwarderAllocateNextHigher(t *testing.T) {
require.True(t, boosted)
// next higher, move from (1, 0) -> (1, 3), still deficient though
f.currentLayers.spatial = 1
f.currentLayers.temporal = 0
f.currentLayers.Spatial = 1
f.currentLayers.Temporal = 0
expectedTargetLayers = VideoLayers{
spatial: 1,
temporal: 3,
Spatial: 1,
Temporal: 3,
}
expectedResult = VideoAllocation{
state: VideoAllocationStateDeficient,
@@ -624,10 +624,10 @@ func TestForwarderAllocateNextHigher(t *testing.T) {
require.True(t, boosted)
// next higher, move from (1, 3) -> (2, 1), optimal allocation
f.currentLayers.temporal = 3
f.currentLayers.Temporal = 3
expectedTargetLayers = VideoLayers{
spatial: 2,
temporal: 1,
Spatial: 2,
Temporal: 1,
}
expectedResult = VideoAllocation{
state: VideoAllocationStateOptimal,
@@ -646,8 +646,8 @@ func TestForwarderAllocateNextHigher(t *testing.T) {
require.True(t, boosted)
// ask again, should return not boosted as there is no room to go higher
f.currentLayers.spatial = 2
f.currentLayers.temporal = 1
f.currentLayers.Spatial = 2
f.currentLayers.Temporal = 1
result, boosted = f.AllocateNextHigher(ChannelCapacityInfinity, bitrates)
require.Equal(t, expectedResult, result)
require.Equal(t, expectedResult, f.lastAllocation)
@@ -660,8 +660,8 @@ func TestForwarderAllocateNextHigher(t *testing.T) {
f.lastAllocation.bandwidthRequested = 0
expectedTargetLayers = VideoLayers{
spatial: 0,
temporal: 0,
Spatial: 0,
Temporal: 0,
}
expectedResult = VideoAllocation{
state: VideoAllocationStateDeficient,
@@ -711,7 +711,7 @@ func TestForwarderPause(t *testing.T) {
f.availableLayers = availableLayers
f.ProvisionalAllocatePrepare(bitrates)
f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}, true)
f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 0, Temporal: 0}, true)
// should have set target at (0, 0)
f.ProvisionalAllocateCommit()
@@ -745,7 +745,7 @@ func TestForwarderPauseMute(t *testing.T) {
f.availableLayers = availableLayers
f.ProvisionalAllocatePrepare(bitrates)
f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}, true)
f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 0, Temporal: 0}, true)
// should have set target at (0, 0)
f.ProvisionalAllocateCommit()
@@ -969,8 +969,8 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) {
// although target layer matches, not a key frame, so should drop and ask to send PLI
f.targetLayers = VideoLayers{
spatial: 0,
temporal: 1,
Spatial: 0,
Temporal: 1,
}
expectedTP = TranslationParams{
shouldDrop: true,
@@ -1222,8 +1222,8 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) {
// switching SSRC (happens for new layer or new track source)
// should lock onto the new source, but sequence number should be contiguous
f.targetLayers = VideoLayers{
spatial: 1,
temporal: 1,
Spatial: 1,
Temporal: 1,
}
params = &testutils.TestExtPacketParams{
@@ -1306,8 +1306,8 @@ func TestForwardGetSnTsForPadding(t *testing.T) {
extPkt, _ := testutils.GetTestExtPacketVP8(params, vp8)
f.targetLayers = VideoLayers{
spatial: 0,
temporal: 1,
Spatial: 0,
Temporal: 1,
}
f.currentLayers = InvalidLayers
@@ -1373,8 +1373,8 @@ func TestForwardGetSnTsForBlankFrames(t *testing.T) {
extPkt, _ := testutils.GetTestExtPacketVP8(params, vp8)
f.targetLayers = VideoLayers{
spatial: 0,
temporal: 1,
Spatial: 0,
Temporal: 1,
}
f.currentLayers = InvalidLayers
@@ -1443,8 +1443,8 @@ func TestForwardGetPaddingVP8(t *testing.T) {
extPkt, _ := testutils.GetTestExtPacketVP8(params, vp8)
f.targetLayers = VideoLayers{
spatial: 0,
temporal: 1,
Spatial: 0,
Temporal: 1,
}
f.currentLayers = InvalidLayers
+39 -8
View File
@@ -3,6 +3,7 @@ package sfu
import (
"errors"
"io"
"strings"
"sync"
"time"
@@ -65,6 +66,7 @@ type WebRTCReceiver struct {
receiver *webrtc.RTPReceiver
codec webrtc.RTPCodecParameters
isSimulcast bool
isSVC bool
onCloseHandler func()
closeOnce sync.Once
closed atomic.Bool
@@ -165,6 +167,13 @@ func NewWebRTCReceiver(
twcc: twcc,
streamTrackerManager: NewStreamTrackerManager(logger, source),
}
switch strings.ToLower(w.codec.MimeType) {
case "video/av1":
fallthrough
case "video/vp9":
w.isSVC = true
}
w.streamTrackerManager.OnAvailableLayersChanged(w.downTrackLayerChange)
w.streamTrackerManager.OnBitrateAvailabilityChanged(w.downTrackBitrateAvailabilityChange)
@@ -381,9 +390,8 @@ func (w *WebRTCReceiver) sendRTCP(packets []rtcp.Packet) {
}
func (w *WebRTCReceiver) SendPLI(layer int32) {
w.bufferMu.RLock()
buff := w.buffers[layer]
w.bufferMu.RUnlock()
// TODO : should send LRR (Layer Refresh Request) instead of PLI
buff := w.getBuffer(layer)
if buff == nil {
return
}
@@ -395,11 +403,23 @@ func (w *WebRTCReceiver) SetRTCPCh(ch chan []rtcp.Packet) {
w.rtcpCh = ch
}
func (w *WebRTCReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) {
func (w *WebRTCReceiver) getBuffer(layer int32) *buffer.Buffer {
// for svc codecs, use layer full quality instead.
// we only have buffer for full quality
if w.isSVC {
layer = int32(len(w.buffers)) - 1
}
w.bufferMu.RLock()
buff := w.buffers[layer]
w.bufferMu.RUnlock()
return buff.GetPacket(buf, sn)
if buff == nil {
w.logger.Warnw("getBuffer failed, buffer not found", nil, "layer", layer)
}
return buff
}
func (w *WebRTCReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) {
return w.getBuffer(int32(layer)).GetPacket(buf, sn)
}
func (w *WebRTCReceiver) GetTrackStats() *livekit.RTPStats {
@@ -532,11 +552,22 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) {
return
}
if tracker != nil {
tracker.Observe(pkt.Packet.SequenceNumber, pkt.TemporalLayer, len(pkt.RawPacket), len(pkt.Packet.Payload))
// svc packet, dispatch to correct tracker
spatialTracker := tracker
spatialLayer := layer
if pkt.Spatial >= 0 {
spatialLayer = pkt.Spatial
spatialTracker = w.streamTrackerManager.GetTracker(pkt.Spatial)
if spatialTracker == nil {
spatialTracker = w.streamTrackerManager.AddTracker(pkt.Spatial)
}
}
w.downTrackSpreader.Broadcast(layer, pkt)
if spatialTracker != nil {
spatialTracker.Observe(pkt.Packet.SequenceNumber, pkt.Temporal, len(pkt.RawPacket), len(pkt.Packet.Payload))
}
w.downTrackSpreader.Broadcast(spatialLayer, pkt)
}
}
+5 -5
View File
@@ -985,8 +985,8 @@ func (s *StreamAllocator) allocateAllTracks() {
for spatial := int32(0); spatial <= DefaultMaxLayerSpatial; spatial++ {
for temporal := int32(0); temporal <= DefaultMaxLayerTemporal; temporal++ {
layers := VideoLayers{
spatial: spatial,
temporal: temporal,
Spatial: spatial,
Temporal: temporal,
}
for _, track := range sorted {
@@ -1424,11 +1424,11 @@ func (t TrackSorter) Less(i, j int) bool {
return t[i].priority > t[j].priority
}
if t[i].maxLayers.spatial != t[j].maxLayers.spatial {
return t[i].maxLayers.spatial > t[j].maxLayers.spatial
if t[i].maxLayers.Spatial != t[j].maxLayers.Spatial {
return t[i].maxLayers.Spatial > t[j].maxLayers.Spatial
}
return t[i].maxLayers.temporal > t[j].maxLayers.temporal
return t[i].maxLayers.Temporal > t[j].maxLayers.Temporal
}
// ------------------------------------------------
+4 -3
View File
@@ -83,17 +83,17 @@ func (s *StreamTrackerManager) OnBitrateAvailabilityChanged(f func()) {
s.onBitrateAvailabilityChanged = f
}
func (s *StreamTrackerManager) AddTracker(layer int32) {
func (s *StreamTrackerManager) AddTracker(layer int32) *StreamTracker {
var params StreamTrackerParams
if s.source == livekit.TrackSource_SCREEN_SHARE {
if int(layer) >= len(ConfigScreenshare) {
return
return nil
}
params = ConfigScreenshare[layer]
} else {
if int(layer) >= len(ConfigVideo) {
return
return nil
}
params = ConfigVideo[layer]
@@ -129,6 +129,7 @@ func (s *StreamTrackerManager) AddTracker(layer int32) {
s.lock.Unlock()
tracker.Start()
return tracker
}
func (s *StreamTrackerManager) RemoveTracker(layer int32) {
+203
View File
@@ -0,0 +1,203 @@
package sfu
import (
"fmt"
"sort"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor"
"github.com/livekit/protocol/logger"
)
type targetLayer struct {
Target int
Layer VideoLayers
}
type DDVideoLayerSelector struct {
logger logger.Logger
// TODO : fields for frame chain detect
// frameNumberWrapper Uint16Wrapper
// expectKeyFrame bool
decodeTargetLayer []targetLayer
layer VideoLayers
activeDecodeTargetsBitmask *uint32
structure *dd.FrameDependencyStructure
}
func NewDDVideoLayerSelector(logger logger.Logger) *DDVideoLayerSelector {
return &DDVideoLayerSelector{
logger: logger,
layer: VideoLayers{Spatial: 2, Temporal: 2},
}
}
func (s *DDVideoLayerSelector) Select(expPkt *buffer.ExtPacket, tp *TranslationParams) (selected bool) {
// return true
tp.marker = expPkt.Packet.Marker
if expPkt.DependencyDescriptor == nil {
// packet don't have dependency descriptor, pass check
return true
}
if expPkt.DependencyDescriptor.AttachedStructure != nil {
// update decode target layer and active decode targets
// TODO : these targets info can be shared by all the downtracks, no need calculate in every selector
s.updateDependencyStructure(expPkt.DependencyDescriptor.AttachedStructure)
}
// forward all packets before locking
if s.layer == InvalidLayers {
return true
}
// 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.
activeDecodeTargets := expPkt.DependencyDescriptor.ActiveDecodeTargetsBitmask
if activeDecodeTargets != nil {
s.logger.Debugw("active decode targets", "activeDecodeTargets", *activeDecodeTargets)
}
currentTarget := -1
for _, dt := range s.decodeTargetLayer {
// find target match with selected layer
if dt.Layer.Spatial <= s.layer.Spatial && dt.Layer.Temporal <= s.layer.Temporal {
if activeDecodeTargets == nil || ((*activeDecodeTargets)&(1<<dt.Target) != 0) {
// TODO : check frame chain integrity
currentTarget = dt.Target
// s.logger.Debugw("select target", "target", currentTarget, "layer", dt.layer, "dtis", expPkt.DependencyDescriptor.FrameDependencies.DecodeTargetIndications)
break
}
}
}
if currentTarget < 0 {
// s.logger.Debugw(fmt.Sprintf("drop packet for no target found, deocdeTargets %v, selected layer %v, s:%d, t:%d",
// s.decodeTargetLayer, s.layer, expPkt.DependencyDescriptor.FrameDependencies.SpatialId, expPkt.DependencyDescriptor.FrameDependencies.TemporalId))
// no active decode target, forward all packets
return false
}
dtis := expPkt.DependencyDescriptor.FrameDependencies.DecodeTargetIndications
if len(dtis) < currentTarget {
// dtis error, dependency descriptor might lost
s.logger.Debugw(fmt.Sprintf("drop packet for dtis error, dtis %v, currentTarget %d, s:%d, t:%d", dtis, currentTarget,
expPkt.DependencyDescriptor.FrameDependencies.SpatialId, expPkt.DependencyDescriptor.FrameDependencies.TemporalId))
return false
}
// TODO : if bandwidth in congest, could drop the 'Discardable' packet
if dti := dtis[currentTarget]; dti == dd.DecodeTargetNotPresent {
// s.logger.Debugw(fmt.Sprintf("drop packet for decode target not present, dtis %v, currentTarget %d, s:%d, t:%d", dtis, currentTarget,
// expPkt.DependencyDescriptor.FrameDependencies.SpatialId, expPkt.DependencyDescriptor.FrameDependencies.TemporalId))
return false
} else if dti == dd.DecodeTargetSwitch {
tp.switchingToTargetLayer = true
}
// TODO : add frame to forwarded queue if entire frame is forwarded
// s.logger.Debugw("select packet", "target", currentTarget, "layer", s.layer)
tp.ddExtension = &dd.DependencyDescriptorExtension{
Descriptor: expPkt.DependencyDescriptor,
Structure: s.structure,
}
if expPkt.DependencyDescriptor.AttachedStructure == nil && s.activeDecodeTargetsBitmask != nil {
// clone and override activebitmask
ddClone := *tp.ddExtension.Descriptor
ddClone.ActiveDecodeTargetsBitmask = s.activeDecodeTargetsBitmask
tp.ddExtension.Descriptor = &ddClone
// s.logger.Debugw("set active decode targets bitmask", "activeDecodeTargetsBitmask", s.activeDecodeTargetsBitmask)
}
mark := expPkt.Packet.Header.Marker || (expPkt.DependencyDescriptor.LastPacketInFrame && s.layer.Spatial == int32(expPkt.DependencyDescriptor.FrameDependencies.SpatialId))
tp.marker = mark
return true
}
func (s *DDVideoLayerSelector) SelectLayer(layer VideoLayers) {
// layer = VideoLayers{1, 1}
s.layer = layer
activeBitMask := uint32(0)
var maxSpatial, maxTemporal int32
for _, dt := range s.decodeTargetLayer {
if dt.Layer.Spatial > maxSpatial {
maxSpatial = dt.Layer.Spatial
}
if dt.Layer.Temporal > maxTemporal {
maxTemporal = dt.Layer.Temporal
}
if dt.Layer.Spatial <= layer.Spatial && dt.Layer.Temporal <= layer.Temporal {
activeBitMask |= 1 << dt.Target
}
}
if layer.Spatial == maxSpatial && layer.Temporal == maxTemporal {
// all the decode targets are selected
s.activeDecodeTargetsBitmask = nil
} else {
s.activeDecodeTargetsBitmask = &activeBitMask
}
s.logger.Debugw("select layer ", "layer", layer, "activeDecodeTargetsBitmask", s.activeDecodeTargetsBitmask)
}
func (s *DDVideoLayerSelector) updateDependencyStructure(structure *dd.FrameDependencyStructure) {
s.structure = structure
s.decodeTargetLayer = s.decodeTargetLayer[:0]
for target := 0; target < structure.NumDecodeTargets; target++ {
layer := VideoLayers{Spatial: 0, Temporal: 0}
for _, t := range structure.Templates {
if t.DecodeTargetIndications[target] != dd.DecodeTargetNotPresent {
if layer.Spatial < int32(t.SpatialId) {
layer.Spatial = int32(t.SpatialId)
}
if layer.Temporal < int32(t.TemporalId) {
layer.Temporal = int32(t.TemporalId)
}
}
}
s.decodeTargetLayer = append(s.decodeTargetLayer, targetLayer{target, layer})
}
// sort decode target layer by spatial and temporal from high to low
sort.Slice(s.decodeTargetLayer, func(i, j int) bool {
if s.decodeTargetLayer[i].Layer.Spatial == s.decodeTargetLayer[j].Layer.Spatial {
return s.decodeTargetLayer[i].Layer.Temporal > s.decodeTargetLayer[j].Layer.Temporal
}
return s.decodeTargetLayer[i].Layer.Spatial > s.decodeTargetLayer[j].Layer.Spatial
})
s.logger.Debugw(fmt.Sprintf("update decode targets: %v", s.decodeTargetLayer))
}
// TODO : use generic wrapper when updated to go 1.18
type Uint16Wrapper struct {
last_value *uint16
lastUnwrapped int32
}
func (w *Uint16Wrapper) Unwrap(value uint16) int32 {
if w.last_value == nil {
w.last_value = &value
w.lastUnwrapped = int32(value)
return int32(*w.last_value)
}
diff := value - *w.last_value
w.lastUnwrapped += int32(diff)
if diff == 0x8000 && value < *w.last_value {
w.lastUnwrapped -= 0x10000
} else if diff > 0x8000 {
w.lastUnwrapped -= 0x10000
}
*w.last_value = value
return w.lastUnwrapped
}