dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 05:04:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Some clean up (#505)

Browse Source 
* WIP commit

* Refactor NTP time

* Clean up

* Update lk protocol
This commit is contained in:
Raja Subramanian
2022-03-11 22:40:49 +05:30
committed by GitHub
parent a5fa54853f
commit 80bd45f061
10 changed files with 82 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -14,7 +14,7 @@ require (
github.com/google/wire v0.5.0
github.com/gorilla/websocket v1.4.2
github.com/hashicorp/golang-lru v0.5.4
github.com/livekit/protocol v0.11.14-0.20220302192533-dbd455d2c1de
github.com/livekit/protocol v0.11.14-0.20220311165704-a91198e400ac
github.com/magefile/mage v1.11.0
github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
github.com/mitchellh/go-homedir v1.1.0
go.sum
+2 -2
View File
@@ -134,8 +134,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA=
github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
github.com/livekit/protocol v0.11.14-0.20220302192533-dbd455d2c1de h1:uyUDLn1HcyxMwbAPX9SVQHfn6O/Zncx7vPH/YXC39wE=
github.com/livekit/protocol v0.11.14-0.20220302192533-dbd455d2c1de/go.mod h1:3pHsWUtQmWaH8mG0cXrQWpbf3Vo+kj0U+In77CEXu90=
github.com/livekit/protocol v0.11.14-0.20220311165704-a91198e400ac h1:86mRiCnqY/wEkKBUO+gYmon0WyCNt1ejVDmxBHteTV8=
github.com/livekit/protocol v0.11.14-0.20220311165704-a91198e400ac/go.mod h1:3pHsWUtQmWaH8mG0cXrQWpbf3Vo+kj0U+In77CEXu90=
github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls=
github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
pkg/rtc/datatrack.go
+2 -1
View File
@@ -5,6 +5,7 @@ import (
"sync"
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/pion/webrtc/v3"
@@ -165,7 +166,7 @@ func (t *DataTrack) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) {
return 0, nil
}
func (t *DataTrack) GetSenderReportTime(layer int32) (rtpTS uint32, ntpTS uint64) {
func (t *DataTrack) GetSenderReportTime(layer int32) (rtpTS uint32, ntpTS buffer.NtpTime) {
return
}
pkg/sfu/buffer/buffer.go
+3 -3
View File
@@ -68,7 +68,7 @@ type Buffer struct {
lastPacketRead int
bitrate atomic.Value
bitrateHelper [4]int64
lastSRNTPTime uint64
lastSRNTPTime NtpTime
lastSRRTPTime uint32
lastSRRecv int64 // Represents wall clock of the most recent sender report arrival
lastTransit uint32
@@ -663,7 +663,7 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport {
func (b *Buffer) SetSenderReportData(rtpTime uint32, ntpTime uint64) {
b.Lock()
b.lastSRRTPTime = rtpTime
b.lastSRNTPTime = ntpTime
b.lastSRNTPTime = NtpTime(ntpTime)
b.lastSRRecv = time.Now().UnixNano()
b.Unlock()
}
@@ -755,7 +755,7 @@ func (b *Buffer) GetClockRate() uint32 {
}
// GetSenderReportData returns the rtp, ntp and nanos of the last sender report
func (b *Buffer) GetSenderReportData() (rtpTime uint32, ntpTime uint64, lastReceivedTimeInNanosSinceEpoch int64) {
func (b *Buffer) GetSenderReportData() (rtpTime uint32, ntpTime NtpTime, lastReceivedTimeInNanosSinceEpoch int64) {
b.RLock()
defer b.RUnlock()
pkg/sfu/buffer/helpers.go
+36
View File
@@ -3,6 +3,7 @@ package buffer
import (
"encoding/binary"
"errors"
"time"
"github.com/livekit/protocol/logger"
)
@@ -275,3 +276,38 @@ func IsH264Keyframe(payload []byte) bool {
}
return false
}
// -------------------------------------
var (
ntpEpoch = time.Date(1900, 1, 1, 0, 0, 0, 0, time.UTC)
)
type NtpTime uint64
func (t NtpTime) Duration() time.Duration {
sec := (t >> 32) * 1e9
frac := (t & 0xffffffff) * 1e9
nsec := frac >> 32
if uint32(frac) >= 0x80000000 {
nsec++
}
return time.Duration(sec + nsec)
}
func (t NtpTime) Time() time.Time {
return ntpEpoch.Add(t.Duration())
}
func ToNtpTime(t time.Time) NtpTime {
nsec := uint64(t.Sub(ntpEpoch))
sec := nsec / 1e9
nsec = (nsec - sec*1e9) << 32
frac := nsec / 1e9
if nsec%1e9 >= 1e9/2 {
frac++
}
return NtpTime(sec<<32 | frac)
}
// ------------------------------------------
pkg/sfu/buffer/helpers_test.go
+31
View File
@@ -2,6 +2,7 @@ package buffer
import (
"testing"
"time"
"github.com/stretchr/testify/require"
)
@@ -92,3 +93,33 @@ func TestVP8Helper_Unmarshal(t *testing.T) {
})
}
}
// ------------------------------------------
func Test_timeToNtp(t *testing.T) {
type args struct {
ns time.Time
}
tests := []struct {
name string
args args
wantNTP uint64
}{
{
name: "Must return correct NTP time",
args: args{
ns: time.Unix(1602391458, 1234),
},
wantNTP: 16369753560730047668,
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
gotNTP := uint64(ToNtpTime(tt.args.ns))
if gotNTP != tt.wantNTP {
t.Errorf("timeToNtp() gotFraction = %v, want %v", gotNTP, tt.wantNTP)
}
})
}
}
pkg/sfu/downtrack.go
+2 -2
View File
@@ -804,9 +804,9 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport {
}
now := time.Now()
nowNTP := toNtpTime(now)
nowNTP := buffer.ToNtpTime(now)
diff := (uint64(now.Sub(ntpTime(srNTP).Time())) * uint64(d.codec.ClockRate)) / uint64(time.Second)
diff := (uint64(now.Sub(srNTP.Time())) * uint64(d.codec.ClockRate)) / uint64(time.Second)
octets, packets := d.getSRStats()
return &rtcp.SenderReport{
pkg/sfu/helpers.go
+3 -31
View File
@@ -5,16 +5,11 @@ import (
"strings"
"time"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/pion/rtcp"
"github.com/pion/webrtc/v3"
)
var (
ntpEpoch = time.Date(1900, 1, 1, 0, 0, 0, 0, time.UTC)
)
type ntpTime uint64
const (
QuarterResolution = "q"
HalfResolution = "h"
@@ -42,30 +37,7 @@ func codecParametersFuzzySearch(needle webrtc.RTPCodecParameters, haystack []web
return webrtc.RTPCodecParameters{}, webrtc.ErrCodecNotFound
}
func (t ntpTime) Duration() time.Duration {
sec := (t >> 32) * 1e9
frac := (t & 0xffffffff) * 1e9
nsec := frac >> 32
if uint32(frac) >= 0x80000000 {
nsec++
}
return time.Duration(sec + nsec)
}
func (t ntpTime) Time() time.Time {
return ntpEpoch.Add(t.Duration())
}
func toNtpTime(t time.Time) ntpTime {
nsec := uint64(t.Sub(ntpEpoch))
sec := nsec / 1e9
nsec = (nsec - sec*1e9) << 32
frac := nsec / 1e9
if nsec%1e9 >= 1e9/2 {
frac++
}
return ntpTime(sec<<32 | frac)
}
// -----------------------------------------------
func getRttMs(report *rtcp.ReceptionReport) uint32 {
if report.LastSenderReport == 0 {
@@ -75,7 +47,7 @@ func getRttMs(report *rtcp.ReceptionReport) uint32 {
// RTT calculation reference: https://datatracker.ietf.org/doc/html/rfc3550#section-6.4.1
// middle 32-bits of current NTP time
now := uint32(toNtpTime(time.Now()) >> 16)
now := uint32(buffer.ToNtpTime(time.Now()) >> 16)
ntpDiff := now - report.LastSenderReport - report.Delay
return uint32(math.Ceil(float64(ntpDiff) * 1000.0 / 65536.0))
}
pkg/sfu/helpers_test.go
-34
View File
@@ -1,34 +0,0 @@
package sfu
import (
"testing"
"time"
)
func Test_timeToNtp(t *testing.T) {
type args struct {
ns time.Time
}
tests := []struct {
name string
args args
wantNTP uint64
}{
{
name: "Must return correct NTP time",
args: args{
ns: time.Unix(1602391458, 1234),
},
wantNTP: 16369753560730047668,
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
gotNTP := uint64(toNtpTime(tt.args.ns))
if gotNTP != tt.wantNTP {
t.Errorf("timeToNtp() gotFraction = %v, want %v", gotNTP, tt.wantNTP)
}
})
}
}
pkg/sfu/receiver.go
+2 -3
View File
@@ -35,11 +35,10 @@ type TrackReceiver interface {
Codec() webrtc.RTPCodecCapability
ReadRTP(buf []byte, layer uint8, sn uint16) (int, error)
GetSenderReportTime(layer int32) (rtpTS uint32, ntpTS uint64)
GetSenderReportTime(layer int32) (rtpTS uint32, ntpTS buffer.NtpTime)
GetBitrateTemporalCumulative() Bitrates
SendPLI(layer int32)
LastPLI() int64
SetUpTrackPaused(paused bool)
SetMaxExpectedSpatialLayer(layer int32)
@@ -424,7 +423,7 @@ func (w *WebRTCReceiver) SetRTCPCh(ch chan []rtcp.Packet) {
w.rtcpCh = ch
}
func (w *WebRTCReceiver) GetSenderReportTime(layer int32) (rtpTS uint32, ntpTS uint64) {
func (w *WebRTCReceiver) GetSenderReportTime(layer int32) (rtpTS uint32, ntpTS buffer.NtpTime) {
w.bufferMu.RLock()
defer w.bufferMu.RUnlock()
if w.buffers[layer] != nil {