Move RTP-specific code into its own package.
This commit is contained in:
Juliusz Chroboczek
+1048
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,74 @@
|
||||
package rtpconn
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"sfu/rtptime"
|
||||
"sfu/stats"
|
||||
)
|
||||
|
||||
func (c *webClient) GetStats() *stats.Client {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
cs := stats.Client{
|
||||
Id: c.id,
|
||||
}
|
||||
|
||||
for _, up := range c.up {
|
||||
conns := stats.Conn{
|
||||
Id: up.id,
|
||||
}
|
||||
tracks := up.getTracks()
|
||||
for _, t := range tracks {
|
||||
expected, lost, _, _ := t.cache.GetStats(false)
|
||||
if expected == 0 {
|
||||
expected = 1
|
||||
}
|
||||
loss := uint8(lost * 100 / expected)
|
||||
jitter := time.Duration(t.jitter.Jitter()) *
|
||||
(time.Second / time.Duration(t.jitter.HZ()))
|
||||
rate, _ := t.rate.Estimate()
|
||||
conns.Tracks = append(conns.Tracks, stats.Track{
|
||||
Bitrate: uint64(rate) * 8,
|
||||
Loss: loss,
|
||||
Jitter: jitter,
|
||||
})
|
||||
}
|
||||
cs.Up = append(cs.Up, conns)
|
||||
}
|
||||
sort.Slice(cs.Up, func(i, j int) bool {
|
||||
return cs.Up[i].Id < cs.Up[j].Id
|
||||
})
|
||||
|
||||
jiffies := rtptime.Jiffies()
|
||||
for _, down := range c.down {
|
||||
conns := stats.Conn{
|
||||
Id: down.id,
|
||||
MaxBitrate: down.GetMaxBitrate(jiffies),
|
||||
}
|
||||
for _, t := range down.tracks {
|
||||
rate, _ := t.rate.Estimate()
|
||||
rtt := rtptime.ToDuration(atomic.LoadUint64(&t.rtt),
|
||||
rtptime.JiffiesPerSec)
|
||||
loss, jitter := t.stats.Get(jiffies)
|
||||
j := time.Duration(jitter) * time.Second /
|
||||
time.Duration(t.track.Codec().ClockRate)
|
||||
conns.Tracks = append(conns.Tracks, stats.Track{
|
||||
Bitrate: uint64(rate) * 8,
|
||||
MaxBitrate: t.maxBitrate.Get(jiffies),
|
||||
Loss: uint8(uint32(loss) * 100 / 256),
|
||||
Rtt: rtt,
|
||||
Jitter: j,
|
||||
})
|
||||
}
|
||||
cs.Down = append(cs.Down, conns)
|
||||
}
|
||||
sort.Slice(cs.Down, func(i, j int) bool {
|
||||
return cs.Down[i].Id < cs.Down[j].Id
|
||||
})
|
||||
|
||||
return &cs
|
||||
}
|
||||
@@ -0,0 +1,304 @@
|
||||
package rtpconn
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/pion/rtp"
|
||||
|
||||
"sfu/conn"
|
||||
"sfu/packetcache"
|
||||
"sfu/rtptime"
|
||||
)
|
||||
|
||||
// packetIndex is a request to send a packet from the cache.
|
||||
type packetIndex struct {
|
||||
// the packet's seqno
|
||||
seqno uint16
|
||||
// the index in the cache
|
||||
index uint16
|
||||
}
|
||||
|
||||
// An rtpWriterPool is a set of rtpWriters
|
||||
type rtpWriterPool struct {
|
||||
conn *rtpUpConnection
|
||||
track *rtpUpTrack
|
||||
writers []*rtpWriter
|
||||
count int
|
||||
}
|
||||
|
||||
// sqrt computes the integer square root
|
||||
func sqrt(n int) int {
|
||||
if n < 2 {
|
||||
return n
|
||||
}
|
||||
|
||||
s := sqrt(n/2) * 2
|
||||
l := s + 1
|
||||
if l*l > n {
|
||||
return s
|
||||
} else {
|
||||
return l
|
||||
}
|
||||
}
|
||||
|
||||
// add adds or removes a track from a writer pool
|
||||
func (wp *rtpWriterPool) add(track conn.DownTrack, add bool) error {
|
||||
n := 4
|
||||
if wp.count > 16 {
|
||||
n = sqrt(wp.count)
|
||||
}
|
||||
|
||||
i := 0
|
||||
for i < len(wp.writers) {
|
||||
w := wp.writers[i]
|
||||
err := w.add(track, add, n)
|
||||
if err == nil {
|
||||
if add {
|
||||
wp.count++
|
||||
} else {
|
||||
if wp.count > 0 {
|
||||
wp.count--
|
||||
} else {
|
||||
log.Printf("Negative writer count!")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
} else if err == ErrWriterDead {
|
||||
wp.del(wp.writers[i])
|
||||
} else {
|
||||
i++
|
||||
}
|
||||
}
|
||||
|
||||
if add {
|
||||
writer := newRtpWriter(wp.conn, wp.track)
|
||||
wp.writers = append(wp.writers, writer)
|
||||
err := writer.add(track, true, n)
|
||||
if err == nil {
|
||||
wp.count++
|
||||
}
|
||||
return err
|
||||
} else {
|
||||
return errors.New("deleting unknown track")
|
||||
}
|
||||
}
|
||||
|
||||
// del deletes a writer.
|
||||
func (wp *rtpWriterPool) del(w *rtpWriter) bool {
|
||||
for i, ww := range wp.writers {
|
||||
if ww == w {
|
||||
close(w.ch)
|
||||
wp.writers = append(wp.writers[:i], wp.writers[i+1:]...)
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// close deletes all writers.
|
||||
func (wp *rtpWriterPool) close() {
|
||||
for _, w := range wp.writers {
|
||||
close(w.ch)
|
||||
}
|
||||
wp.writers = nil
|
||||
wp.count = 0
|
||||
}
|
||||
|
||||
// write writes a packet stored in the packet cache to all local tracks
|
||||
func (wp *rtpWriterPool) write(seqno uint16, index uint16, delay uint32, isvideo bool, marker bool) {
|
||||
pi := packetIndex{seqno, index}
|
||||
|
||||
var dead []*rtpWriter
|
||||
for _, w := range wp.writers {
|
||||
if w.drop > 0 {
|
||||
// currently dropping
|
||||
if marker {
|
||||
// last packet in frame
|
||||
w.drop = 0
|
||||
} else {
|
||||
w.drop--
|
||||
}
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case w.ch <- pi:
|
||||
// all is well
|
||||
case <-w.done:
|
||||
// the writer is dead.
|
||||
dead = append(dead, w)
|
||||
default:
|
||||
// the writer is congested
|
||||
if isvideo {
|
||||
// drop until the end of the frame
|
||||
if !marker {
|
||||
w.drop = 7
|
||||
}
|
||||
continue
|
||||
}
|
||||
// audio, try again with a delay
|
||||
d := delay/uint32(2*len(wp.writers))
|
||||
timer := time.NewTimer(rtptime.ToDuration(
|
||||
uint64(d), rtptime.JiffiesPerSec,
|
||||
))
|
||||
|
||||
select {
|
||||
case w.ch <- pi:
|
||||
timer.Stop()
|
||||
case <-w.done:
|
||||
dead = append(dead, w)
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if dead != nil {
|
||||
for _, d := range dead {
|
||||
wp.del(d)
|
||||
}
|
||||
dead = nil
|
||||
}
|
||||
}
|
||||
|
||||
var ErrWriterDead = errors.New("writer is dead")
|
||||
var ErrWriterBusy = errors.New("writer is busy")
|
||||
var ErrUnknownTrack = errors.New("unknown track")
|
||||
|
||||
type writerAction struct {
|
||||
add bool
|
||||
track conn.DownTrack
|
||||
maxTracks int
|
||||
ch chan error
|
||||
}
|
||||
|
||||
// an rtpWriter is a thread writing to a set of tracks.
|
||||
type rtpWriter struct {
|
||||
ch chan packetIndex
|
||||
done chan struct{}
|
||||
action chan writerAction
|
||||
|
||||
// this is not touched by the writer loop, used by the caller
|
||||
drop int
|
||||
}
|
||||
|
||||
func newRtpWriter(conn *rtpUpConnection, track *rtpUpTrack) *rtpWriter {
|
||||
writer := &rtpWriter{
|
||||
ch: make(chan packetIndex, 32),
|
||||
done: make(chan struct{}),
|
||||
action: make(chan writerAction, 1),
|
||||
}
|
||||
go rtpWriterLoop(writer, conn, track)
|
||||
return writer
|
||||
}
|
||||
|
||||
// add adds or removes a track from a writer.
|
||||
func (writer *rtpWriter) add(track conn.DownTrack, add bool, max int) error {
|
||||
ch := make(chan error, 1)
|
||||
select {
|
||||
case writer.action <- writerAction{add, track, max, ch}:
|
||||
select {
|
||||
case err := <-ch:
|
||||
return err
|
||||
case <-writer.done:
|
||||
return ErrWriterDead
|
||||
}
|
||||
case <-writer.done:
|
||||
return ErrWriterDead
|
||||
}
|
||||
}
|
||||
|
||||
// rtpWriterLoop is the main loop of an rtpWriter.
|
||||
func rtpWriterLoop(writer *rtpWriter, up *rtpUpConnection, track *rtpUpTrack) {
|
||||
defer close(writer.done)
|
||||
|
||||
buf := make([]byte, packetcache.BufSize)
|
||||
var packet rtp.Packet
|
||||
|
||||
local := make([]conn.DownTrack, 0)
|
||||
|
||||
// reset whenever a new track is inserted
|
||||
firSent := false
|
||||
|
||||
for {
|
||||
select {
|
||||
case action := <-writer.action:
|
||||
if action.add {
|
||||
if len(local) >= action.maxTracks {
|
||||
action.ch <- ErrWriterBusy
|
||||
close(action.ch)
|
||||
continue
|
||||
}
|
||||
local = append(local, action.track)
|
||||
action.ch <- nil
|
||||
close(action.ch)
|
||||
|
||||
firSent = false
|
||||
track.mu.Lock()
|
||||
ntp := track.srNTPTime
|
||||
rtp := track.srRTPTime
|
||||
cname := track.cname
|
||||
track.mu.Unlock()
|
||||
if ntp != 0 {
|
||||
action.track.SetTimeOffset(ntp, rtp)
|
||||
}
|
||||
if cname != "" {
|
||||
action.track.SetCname(cname)
|
||||
}
|
||||
} else {
|
||||
found := false
|
||||
for i, t := range local {
|
||||
if t == action.track {
|
||||
local = append(local[:i],
|
||||
local[i+1:]...)
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
action.ch <- ErrUnknownTrack
|
||||
} else {
|
||||
action.ch <- nil
|
||||
}
|
||||
close(action.ch)
|
||||
if len(local) == 0 {
|
||||
return
|
||||
}
|
||||
}
|
||||
case pi, ok := <-writer.ch:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
bytes := track.cache.GetAt(pi.seqno, pi.index, buf)
|
||||
if bytes == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
err := packet.Unmarshal(buf[:bytes])
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
kfNeeded := false
|
||||
for _, l := range local {
|
||||
err := l.WriteRTP(&packet)
|
||||
if err != nil {
|
||||
if err == conn.ErrKeyframeNeeded {
|
||||
kfNeeded = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
l.Accumulate(uint32(bytes))
|
||||
}
|
||||
|
||||
if kfNeeded {
|
||||
err := up.sendFIR(track, !firSent)
|
||||
if err == ErrUnsupportedFeedback {
|
||||
up.sendPLI(track)
|
||||
}
|
||||
firSent = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user