diff --git a/go.mod b/go.mod index 33918dd42..42c50f1e4 100644 --- a/go.mod +++ b/go.mod @@ -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 +) diff --git a/go.sum b/go.sum index dc24c4419..9c163700b 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index 29d8b12c3..7046da6ed 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -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"}, diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index 61da360be..493411d86 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -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) +} diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index ff5c4e675..403b5986d 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -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 +} diff --git a/pkg/sfu/buffer/dependencydescriptorparser.go b/pkg/sfu/buffer/dependencydescriptorparser.go new file mode 100644 index 000000000..77cf98ef9 --- /dev/null +++ b/pkg/sfu/buffer/dependencydescriptorparser.go @@ -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< 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 +} diff --git a/pkg/sfu/dependencydescriptor/bitstreamreader.go b/pkg/sfu/dependencydescriptor/bitstreamreader.go new file mode 100644 index 000000000..84700f6b8 --- /dev/null +++ b/pkg/sfu/dependencydescriptor/bitstreamreader.go @@ -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 +} diff --git a/pkg/sfu/dependencydescriptor/bitstreamwriter.go b/pkg/sfu/dependencydescriptor/bitstreamwriter.go new file mode 100644 index 000000000..671fcdc50 --- /dev/null +++ b/pkg/sfu/dependencydescriptor/bitstreamwriter.go @@ -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 + } +} diff --git a/pkg/sfu/dependencydescriptor/dependencydescriptorextension.go b/pkg/sfu/dependencydescriptor/dependencydescriptorextension.go new file mode 100644 index 000000000..8316b6b01 --- /dev/null +++ b/pkg/sfu/dependencydescriptor/dependencydescriptorextension.go @@ -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 +} diff --git a/pkg/sfu/dependencydescriptor/dependencydescriptorextension_test.go b/pkg/sfu/dependencydescriptor/dependencydescriptorextension_test.go new file mode 100644 index 000000000..25cb1b3bc --- /dev/null +++ b/pkg/sfu/dependencydescriptor/dependencydescriptorextension_test.go @@ -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()) + } +} diff --git a/pkg/sfu/dependencydescriptor/dependencydescriptorreader.go b/pkg/sfu/dependencydescriptor/dependencydescriptorreader.go new file mode 100644 index 000000000..fe69af5e4 --- /dev/null +++ b/pkg/sfu/dependencydescriptor/dependencydescriptorreader.go @@ -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 +} diff --git a/pkg/sfu/dependencydescriptor/dependencydescriptorwriter.go b/pkg/sfu/dependencydescriptor/dependencydescriptorwriter.go new file mode 100644 index 000000000..0f6972898 --- /dev/null +++ b/pkg/sfu/dependencydescriptor/dependencydescriptorwriter.go @@ -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< 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< 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 +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index e22d71155..10c4457c9 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -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, } } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 9936582f1..b315f67e5 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -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 diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 0e37150f8..f35b1a4e7 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -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 diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 2fdaa4c99..83f459345 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -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) } } diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index 39200c28e..e72aa6e13 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -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 } // ------------------------------------------------ diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index f325d9454..51f2bcb08 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -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) { diff --git a/pkg/sfu/videolayerselector.go b/pkg/sfu/videolayerselector.go new file mode 100644 index 000000000..024b64d86 --- /dev/null +++ b/pkg/sfu/videolayerselector.go @@ -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< 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 +}