From 48a9b10ce806a319aa0efc5156048bb31d8a12d9 Mon Sep 17 00:00:00 2001 From: Juliusz Chroboczek Date: Tue, 3 Aug 2021 13:10:55 +0200 Subject: [PATCH] Use spatial scalability when simulcast is not available. If the client requested a low-resolution stream and there is none, switch to SID=0. --- rtpconn/rtpconn.go | 25 ++++++++++++++++++++++--- rtpconn/rtpconn_test.go | 2 +- rtpconn/webclient.go | 28 ++++++++++++++++++++++------ 3 files changed, 45 insertions(+), 10 deletions(-) diff --git a/rtpconn/rtpconn.go b/rtpconn/rtpconn.go index 83f0605..1df9c1f 100644 --- a/rtpconn/rtpconn.go +++ b/rtpconn/rtpconn.go @@ -129,8 +129,12 @@ func (down *rtpDownTrack) SetCname(cname string) { } type layerInfo struct { + // current sid, desired sid, and max sid seen sid, wantedSid, maxSid uint8 + // current tid, desired tid, and max tid seen tid, wantedTid, maxTid uint8 + // if true, stick to sid 0 + limitSid bool } func (down *rtpDownTrack) getLayerInfo() layerInfo { @@ -139,6 +143,7 @@ func (down *rtpDownTrack) getLayerInfo() layerInfo { sid: uint8((info & 0xF)), wantedSid: uint8((info >> 4) & 0xF), maxSid: uint8((info >> 8) & 0xF), + limitSid: ((info >> 12) & 1) != 0, tid: uint8((info >> 16) & 0xF), wantedTid: uint8((info >> 20) & 0xF), maxTid: uint8((info >> 24) & 0xF), @@ -146,10 +151,15 @@ func (down *rtpDownTrack) getLayerInfo() layerInfo { } func (down *rtpDownTrack) setLayerInfo(info layerInfo) { + var l uint32 + if info.limitSid { + l = 1 << 12 + } atomic.StoreUint32(&down.atomics.layerInfo, uint32(info.sid&0xF)| uint32(info.wantedSid&0xF)<<4| uint32(info.maxSid&0xF)<<8| + l| uint32(info.tid&0xF)<<16| uint32(info.wantedTid&0xF)<<20| uint32(info.maxTid&0xF)<<24, @@ -221,6 +231,7 @@ func (down *rtpDownTrack) Write(buf []byte) (int, error) { layer := down.getLayerInfo() + // increase eagerly if this is the first time we see a given layer if flags.Tid > layer.maxTid || flags.Sid > layer.maxSid { if flags.Tid > layer.maxTid { if layer.tid == layer.maxTid { @@ -230,7 +241,7 @@ func (down *rtpDownTrack) Write(buf []byte) (int, error) { layer.maxTid = flags.Tid } if flags.Sid > layer.maxSid { - if layer.sid == layer.maxSid { + if layer.sid == layer.maxSid && !layer.limitSid { layer.wantedSid = flags.Sid layer.sid = flags.Sid } @@ -240,6 +251,7 @@ func (down *rtpDownTrack) Write(buf []byte) (int, error) { down.adjustLayer() layer = down.getLayerInfo() } + if flags.Start && (layer.tid != layer.wantedTid) { if layer.wantedTid < layer.tid || flags.TidUpSync { layer.tid = layer.wantedTid @@ -331,7 +343,10 @@ func (t *rtpDownTrack) adjustLayer() { if rate < max*7/8 { // switch up layer := t.getLayerInfo() - if layer.sid < layer.maxSid { + if layer.limitSid && layer.wantedSid != 0 { + layer.wantedSid = 0 + t.setLayerInfo(layer) + } else if !layer.limitSid && layer.sid < layer.maxSid { layer.wantedSid = layer.sid + 1 t.setLayerInfo(layer) } else if layer.tid < layer.maxTid { @@ -345,7 +360,11 @@ func (t *rtpDownTrack) adjustLayer() { layer.wantedTid = layer.tid - 1 t.setLayerInfo(layer) } else if layer.sid > 0 { - layer.wantedSid = layer.sid - 1 + if layer.limitSid { + layer.wantedSid = 0 + } else { + layer.wantedSid = layer.sid - 1 + } t.setLayerInfo(layer) } } diff --git a/rtpconn/rtpconn_test.go b/rtpconn/rtpconn_test.go index 4cbb2c8..179eb3a 100644 --- a/rtpconn/rtpconn_test.go +++ b/rtpconn/rtpconn_test.go @@ -18,7 +18,7 @@ func TestDownTrackAtomics(t *testing.T) { down.setSRTime(4, 5) down.maxBitrate.Set(6, rtptime.Jiffies()) down.maxREMBBitrate.Set(7, rtptime.Jiffies()) - info := layerInfo{8, 9, 10, 11, 12, 13} + info := layerInfo{8, 9, 10, 11, 12, 13, true} down.setLayerInfo(info) ntp, rtp := down.getTimeOffset() rtt := down.getRTT() diff --git a/rtpconn/webclient.go b/rtpconn/webclient.go index ebe3ea7..4147638 100644 --- a/rtpconn/webclient.go +++ b/rtpconn/webclient.go @@ -446,7 +446,7 @@ func delDownTrackUnlocked(conn *rtpDownConnection, track *rtpDownTrack) error { return os.ErrNotExist } -func replaceTracks(conn *rtpDownConnection, remote []conn.UpTrack) (bool, error) { +func replaceTracks(conn *rtpDownConnection, remote []conn.UpTrack, limitSid bool) (bool, error) { conn.mu.Lock() defer conn.mu.Unlock() @@ -489,6 +489,17 @@ outer2: del = append(del, track) } + defer func() { + for _, t := range conn.tracks { + layer := t.getLayerInfo() + layer.limitSid = limitSid + if limitSid { + layer.wantedSid = 0 + } + t.setLayerInfo(layer) + } + }() + if len(del) == 0 && len(add) == 0 { return false, nil } @@ -729,7 +740,7 @@ func requestConns(target group.Client, g *group.Group, id string) { } } -func requestedTracks(c *webClient, override []string, up conn.Up, tracks []conn.UpTrack) []conn.UpTrack { +func requestedTracks(c *webClient, override []string, up conn.Up, tracks []conn.UpTrack) ([]conn.UpTrack, bool) { r := override if r == nil { var ok bool @@ -738,7 +749,7 @@ func requestedTracks(c *webClient, override []string, up conn.Up, tracks []conn. r, ok = c.requested[""] } if !ok || len(r) == 0 { - return nil + return nil, false } } @@ -777,6 +788,7 @@ func requestedTracks(c *webClient, override []string, up conn.Up, tracks []conn. } var ts []conn.UpTrack + limitSid := false if audio { t := find(webrtc.RTPCodecTypeAudio) if t != nil { @@ -796,10 +808,13 @@ func requestedTracks(c *webClient, override []string, up conn.Up, tracks []conn. ) if t != nil { ts = append(ts, t) + if t.Label() != "l" { + limitSid = true + } } } - return ts + return ts, limitSid } func (c *webClient) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack, replace string) error { @@ -991,6 +1006,7 @@ func clientLoop(c *webClient, ws *websocket.Conn) error { func pushDownConn(c *webClient, id string, up conn.Up, tracks []conn.UpTrack, replace string) error { var requested []conn.UpTrack + limitSid := false if up != nil { var old *rtpDownConnection if replace != "" { @@ -1002,7 +1018,7 @@ func pushDownConn(c *webClient, id string, up conn.Up, tracks []conn.UpTrack, re if old != nil { override = old.requested } - requested = requestedTracks(c, override, up, tracks) + requested, limitSid = requestedTracks(c, override, up, tracks) } if replace != "" { @@ -1031,7 +1047,7 @@ func pushDownConn(c *webClient, id string, up conn.Up, tracks []conn.UpTrack, re } return err } - done, err := replaceTracks(down, requested) + done, err := replaceTracks(down, requested, limitSid) if err != nil || !done { return err }