1
Fork 0
mirror of https://github.com/jech/galene.git synced 2024-11-14 04:35:57 +01:00

Implement packet pacing.

We used to send all of the packets corresponding to a received packet
as a single burst, which led to persistent packet drops for clients
at the end of the queue.  Pace the writer, which introduces up to 1ms
extra jitter.
This commit is contained in:
Juliusz Chroboczek 2020-06-09 14:52:53 +02:00
parent 2454e33df3
commit 4a526b4133

View file

@ -480,6 +480,7 @@ func newUpConn(c client, id string) (*rtpUpConnection, error) {
type packetIndex struct { type packetIndex struct {
seqno uint16 seqno uint16
index uint16 index uint16
delay uint32
} }
func readLoop(conn *rtpUpConnection, track *rtpUpTrack) { func readLoop(conn *rtpUpConnection, track *rtpUpTrack) {
@ -531,15 +532,34 @@ func readLoop(conn *rtpUpConnection, track *rtpUpTrack) {
continue continue
} }
_, rate := track.rate.Estimate()
delay := uint32(rtptime.JiffiesPerSec / 1024)
if rate > 512 {
delay = rtptime.JiffiesPerSec / rate / 2
}
pi := packetIndex{packet.SequenceNumber, index, delay}
select { select {
case ch <- packetIndex{packet.SequenceNumber, index}: case ch <- pi:
default: default:
// the writer is congested
if isvideo { if isvideo {
// the writer is congested. Drop until // keep dropping until the end of the frame
// the end of the frame.
if isvideo && !packet.Marker { if isvideo && !packet.Marker {
drop = 7 drop = 7
} }
} else {
// try again with half the delay on our side
timer := time.NewTimer(rtptime.ToDuration(
uint64(delay/2),
rtptime.JiffiesPerSec,
))
pi.delay = delay / 2
select {
case ch <- pi:
timer.Stop()
case <-timer.C:
}
} }
} }
} }
@ -598,8 +618,15 @@ func writeLoop(conn *rtpUpConnection, track *rtpUpTrack, ch <-chan packetIndex)
continue continue
} }
kfNeeded := false var delay time.Duration
if len(local) > 0 {
delay = rtptime.ToDuration(
uint64(pi.delay / uint32(len(local))),
rtptime.JiffiesPerSec,
)
}
kfNeeded := false
for _, l := range local { for _, l := range local {
err := l.WriteRTP(&packet) err := l.WriteRTP(&packet)
if err != nil { if err != nil {
@ -611,6 +638,9 @@ func writeLoop(conn *rtpUpConnection, track *rtpUpTrack, ch <-chan packetIndex)
continue continue
} }
l.Accumulate(uint32(bytes)) l.Accumulate(uint32(bytes))
if delay > 0 {
time.Sleep(delay)
}
} }
if kfNeeded { if kfNeeded {