Scalable video coding (SVC).

2021-05-11 15:28:30 +02:00
parent 7590588a54
commit 6f9d7fc306
13 changed files with 660 additions and 39 deletions
@@ -1,6 +1,7 @@
 package rtpconn

 import (
+	"errors"
 	"strings"

 	"github.com/pion/rtp"
@@ -108,3 +109,101 @@ func isKeyframe(codec string, packet *rtp.Packet) (bool, bool) {
 		return false, false
 	}
 }
+
+var errTruncated = errors.New("truncated packet")
+var errUnsupportedCodec = errors.New("unsupported codec")
+
+func packetFlags(codec string, buf []byte) (seqno uint16, start bool, pid uint16, tid uint8, sid uint8, layersync bool, discardable bool, err error) {
+	if len(buf) < 12 {
+		err = errTruncated
+		return
+	}
+
+	seqno = (uint16(buf[2]) << 8) | uint16(buf[3])
+
+	if strings.EqualFold(codec, "video/vp8") {
+		var packet rtp.Packet
+		err = packet.Unmarshal(buf)
+		if err != nil {
+			return
+		}
+		var vp8 codecs.VP8Packet
+		_, err = vp8.Unmarshal(packet.Payload)
+		if err != nil {
+			return
+		}
+
+		start = vp8.S == 1 && vp8.PID == 0
+		pid = vp8.PictureID
+		tid = vp8.TID
+		layersync = vp8.Y == 1
+		discardable = vp8.N == 1
+		return
+	} else if strings.EqualFold(codec, "video/vp9") {
+		var packet rtp.Packet
+		err = packet.Unmarshal(buf)
+		if err != nil {
+			return
+		}
+		var vp9 codecs.VP9Packet
+		_, err = vp9.Unmarshal(packet.Payload)
+		if err != nil {
+			return
+		}
+		start = vp9.B
+		tid = vp9.TID
+		sid = vp9.SID
+		layersync = vp9.U
+		return
+	}
+	return
+}
+
+func rewritePacket(codec string, data []byte, seqno uint16, delta uint16) error {
+	if len(data) < 12 {
+		return errTruncated
+	}
+
+	data[2] = uint8(seqno >> 8)
+	data[3] = uint8(seqno)
+	if delta == 0 {
+		return nil
+	}
+
+	offset := 12
+	offset += int(data[0]&0x0F) * 4
+	if len(data) < offset+4 {
+		return errTruncated
+	}
+
+	if (data[0] & 0x10) != 0 {
+		length := uint16(data[offset+2])<<8 | uint16(data[offset+3])
+		offset += 4 + int(length)*4
+		if len(data) < offset+4 {
+			return errTruncated
+		}
+	}
+
+	if strings.EqualFold(codec, "video/vp8") {
+		x := (data[offset] & 0x80) != 0
+		if !x {
+			return nil
+		}
+		i := (data[offset+1] & 0x80) != 0
+		if !i {
+			return nil
+		}
+		m := (data[offset+2] & 0x80) != 0
+		if m {
+			pid := (uint16(data[offset+2]&0x7F) << 8) |
+				uint16(data[offset+3])
+			pid = (pid + delta) & 0x7FFF
+			data[offset+2] = 0x80 | byte((pid>>8)&0x7F)
+			data[offset+3] = byte(pid & 0xFF)
+		} else {
+			data[offset+2] = (data[offset+2] + uint8(delta)) & 0x7F
+		}
+		return nil
+	}
+	return errUnsupportedCodec
+}
@@ -20,6 +20,7 @@ import (
 	"github.com/jech/galene/ice"
 	"github.com/jech/galene/jitter"
 	"github.com/jech/galene/packetcache"
+	"github.com/jech/galene/packetmap"
 	"github.com/jech/galene/rtptime"
 )

@@ -74,6 +75,7 @@ type downTrackAtomics struct {
 	srNTP     uint64
 	remoteNTP uint64
 	remoteRTP uint32
+	layerInfo uint32
 }

 type rtpDownTrack struct {
@@ -81,6 +83,7 @@ type rtpDownTrack struct {
 	sender         *webrtc.RTPSender
 	remote         conn.UpTrack
 	ssrc           webrtc.SSRC
+	packetmap      packetmap.Map
 	maxBitrate     *bitrate
 	maxREMBBitrate *bitrate
 	rate           *estimator.Estimator
@@ -89,14 +92,6 @@ type rtpDownTrack struct {
 	cname          atomic.Value
 }

-func (down *rtpDownTrack) Write(buf []byte) (int, error) {
-	n, err := down.track.Write(buf)
-	if err == nil {
-		down.rate.Accumulate(uint32(n))
-	}
-	return n, err
-}
-
 func (down *rtpDownTrack) SetTimeOffset(ntp uint64, rtp uint32) {
 	atomic.StoreUint64(&down.atomics.remoteNTP, ntp)
 	atomic.StoreUint32(&down.atomics.remoteRTP, rtp)
@@ -131,6 +126,17 @@ func (down *rtpDownTrack) SetCname(cname string) {
 	down.cname.Store(cname)
 }

+func (down *rtpDownTrack) getLayerInfo() (uint8, uint8, uint8) {
+	info := atomic.LoadUint32(&down.atomics.layerInfo)
+	return uint8(info >> 16), uint8(info >> 8), uint8(info)
+}
+
+func (down *rtpDownTrack) setLayerInfo(layer, wanted, max uint8) {
+	atomic.StoreUint32(&down.atomics.layerInfo,
+		(uint32(layer)<<16)|(uint32(wanted)<<8)|uint32(max),
+	)
+}
+
 const (
 	negotiationUnneeded = iota
 	negotiationNeeded
@@ -179,17 +185,108 @@ func newDownConn(c group.Client, id string, remote conn.Up) (*rtpDownConnection,
 	return conn, nil
 }

-func (t *rtpDownTrack) GetMaxBitrate() uint64 {
+var packetBufPool = sync.Pool{
+	New: func() interface{} {
+		return make([]byte, packetcache.BufSize)
+	},
+}
+
+func (down *rtpDownTrack) Write(buf []byte) (int, error) {
+	codec := down.remote.Codec().MimeType
+
+	seqno, start, pid, tid, _, u, _, err := packetFlags(codec, buf)
+	if err != nil {
+		return 0, err
+	}
+
+	layer, wantedLayer, maxLayer := down.getLayerInfo()
+
+	if tid > maxLayer {
+		if layer == maxLayer {
+			wantedLayer = tid
+			layer = tid
+		}
+		maxLayer = tid
+		if wantedLayer > maxLayer {
+			wantedLayer = maxLayer
+		}
+		down.setLayerInfo(layer, wantedLayer, maxLayer)
+		down.adjustLayer()
+	}
+	if start && layer != wantedLayer {
+		if u || wantedLayer < layer {
+			layer = wantedLayer
+			down.setLayerInfo(layer, wantedLayer, maxLayer)
+		}
+	}
+
+	if tid > layer {
+		ok := down.packetmap.Drop(seqno, pid)
+		if ok {
+			return 0, nil
+		}
+	}
+
+	ok, newseqno, piddelta := down.packetmap.Map(seqno, pid)
+	if !ok {
+		return 0, nil
+	}
+
+	if newseqno == seqno && piddelta == 0 {
+		return down.write(buf)
+	}
+
+	ibuf2 := packetBufPool.Get()
+	defer packetBufPool.Put(ibuf2)
+	buf2 := ibuf2.([]byte)
+
+	n := copy(buf2, buf)
+	err = rewritePacket(codec, buf2[:n], newseqno, piddelta)
+	if err != nil {
+		return 0, err
+	}
+	return down.write(buf2[:n])
+}
+
+func (down *rtpDownTrack) write(buf []byte) (int, error) {
+	n, err := down.track.Write(buf)
+	if err == nil {
+		down.rate.Accumulate(uint32(n))
+	}
+	return n, err
+}
+
+func (t *rtpDownTrack) GetMaxBitrate() (uint64, int) {
 	now := rtptime.Jiffies()
+	layer, _, _ := t.getLayerInfo()
 	r := t.maxBitrate.Get(now)
 	if r == ^uint64(0) {
 		r = 512 * 1024
 	}
 	rr := t.maxREMBBitrate.Get(now)
 	if rr == 0 || r < rr {
-		return r
+		return r, int(layer)
+	}
+	return rr, int(layer)
+}
+
+func (t *rtpDownTrack) adjustLayer() {
+	max, _ := t.GetMaxBitrate()
+	r, _ := t.rate.Estimate()
+	rate := uint64(r) * 8
+	if rate < max*7/8 {
+		layer, wanted, max := t.getLayerInfo()
+		if layer < max {
+			wanted = layer + 1
+			t.setLayerInfo(layer, wanted, max)
+		}
+	} else if rate > max*3/2 {
+		layer, wanted, max := t.getLayerInfo()
+		if layer > 0 {
+			wanted = layer - 1
+			t.setLayerInfo(layer, wanted, max)
+		}
 	}
-	return rr
 }

 func (down *rtpDownConnection) addICECandidate(candidate *webrtc.ICECandidateInit) error {
@@ -240,6 +337,7 @@ type rtpUpTrack struct {
 	srTime        uint64
 	srNTPTime     uint64
 	srRTPTime     uint32
+	maxLayer      uint8
 	local         []conn.DownTrack
 	bufferedNACKs []uint16
 }
@@ -598,7 +696,11 @@ func gotNACK(conn *rtpDownConnection, track *rtpDownTrack, p *rtcp.TransportLaye
 	var packet rtp.Packet
 	buf := make([]byte, packetcache.BufSize)
 	for _, nack := range p.Nacks {
-		nack.Range(func(seqno uint16) bool {
+		nack.Range(func(s uint16) bool {
+			ok, seqno, _ := track.packetmap.Reverse(s)
+			if !ok {
+				return true
+			}
 			l := track.remote.GetRTP(seqno, buf)
 			if l == 0 {
 				unhandled = append(unhandled, seqno)
@@ -785,28 +887,44 @@ func sendUpRTCP(up *rtpUpConnection) error {
 			continue
 		}
 		ssrcs = append(ssrcs, uint32(t.track.SSRC()))
-		var r uint64
 		if t.Kind() == webrtc.RTPCodecTypeAudio {
-			r = 100 * 1024
+			rate += 100 * 1024
 		} else if t.Label() == "l" {
-			r = group.LowBitrate
+			rate += group.LowBitrate
 		} else {
+			minrate := ^uint64(0)
+			maxrate := uint64(group.MinBitrate)
+			maxlayer := 0
 			local := t.getLocal()
-			r = ^uint64(0)
 			for _, down := range local {
-				rr := down.GetMaxBitrate()
-				if rr < group.MinBitrate {
-					rr = group.MinBitrate
+				r, l := down.GetMaxBitrate()
+				if maxlayer < l {
+					maxlayer = l
 				}
-				if r > rr {
-					r = rr
+				if r < group.MinBitrate {
+					r = group.MinBitrate
+				}
+				if minrate > r {
+					minrate = r
+				}
+				if maxrate < r {
+					maxrate = r
 				}
 			}
-			if r == ^uint64(0) {
-				r = 512 * 1024
+			// assume that each layer takes two times less
+			// throughput than the higher one.  Then we've
+			// got enough slack for a factor of 2^(layers-1).
+			for i := 0; i < maxlayer; i++ {
+				if minrate < ^uint64(0)/2 {
+					minrate *= 2
+				}
+			}
+			if minrate < maxrate {
+				rate += minrate
+			} else {
+				rate += maxrate
 			}
 		}
-		rate += r
 	}

 	if rate < ^uint64(0) && len(ssrcs) > 0 {
@@ -968,6 +1086,7 @@ func rtcpDownListener(conn *rtpDownConnection, track *rtpDownTrack, s *webrtc.RT
 			continue
 		}

+		adjust := false
 		jiffies := rtptime.Jiffies()

 		for _, p := range ps {
@@ -994,10 +1113,12 @@ func rtcpDownListener(conn *rtpDownConnection, track *rtpDownTrack, s *webrtc.RT
 				}
 			case *rtcp.ReceiverEstimatedMaximumBitrate:
 				track.maxREMBBitrate.Set(p.Bitrate, jiffies)
+				adjust = true
 			case *rtcp.ReceiverReport:
 				for _, r := range p.Reports {
 					if r.SSRC == uint32(track.ssrc) {
 						handleReport(track, r, jiffies)
+						adjust = true
 					}
 				}
 			case *rtcp.SenderReport:
@@ -1010,6 +1131,9 @@ func rtcpDownListener(conn *rtpDownConnection, track *rtpDownTrack, s *webrtc.RT
 				gotNACK(conn, track, p)
 			}
 		}
+		if adjust {
+			track.adjustLayer()
+		}
 	}
 }

@@ -46,9 +46,11 @@ func (c *webClient) GetStats() *stats.Client {
 	jiffies := rtptime.Jiffies()
 	for _, down := range c.down {
 		conns := stats.Conn{
-			Id:         down.id,
+			Id: down.id,
 		}
 		for _, t := range down.tracks {
+			l, _, _ := t.getLayerInfo()
+			layer := int(l)
 			rate, _ := t.rate.Estimate()
 			rtt := rtptime.ToDuration(t.getRTT(),
 				rtptime.JiffiesPerSec)
@@ -56,6 +58,7 @@ func (c *webClient) GetStats() *stats.Client {
 			j := time.Duration(jitter) * time.Second /
 				time.Duration(t.track.Codec().ClockRate)
 			conns.Tracks = append(conns.Tracks, stats.Track{
+				Layer:      &layer,
 				Bitrate:    uint64(rate) * 8,
 				MaxBitrate: t.maxBitrate.Get(jiffies),
 				Loss:       float64(loss) / 256.0,
@@ -1270,6 +1270,9 @@ func handleClientMessage(c *webClient, m clientMessage) error {
 			return closeDownConn(c, m.Id, message)
 		}
 		down := getDownConn(c, m.Id)
+		if down == nil {
+			return ErrUnknownId
+		}
 		if down.negotiationNeeded > negotiationUnneeded {
 			err := negotiate(
 				c, down,