From 80bd45f06170e9506fb43e72b905669826c42308 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 11 Mar 2022 22:40:49 +0530 Subject: [PATCH] Some clean up (#505) * WIP commit * Refactor NTP time * Clean up * Update lk protocol --- go.mod | 2 +- go.sum | 4 ++-- pkg/rtc/datatrack.go | 3 ++- pkg/sfu/buffer/buffer.go | 6 +++--- pkg/sfu/buffer/helpers.go | 36 ++++++++++++++++++++++++++++++++++ pkg/sfu/buffer/helpers_test.go | 31 +++++++++++++++++++++++++++++ pkg/sfu/downtrack.go | 4 ++-- pkg/sfu/helpers.go | 34 +++----------------------------- pkg/sfu/helpers_test.go | 34 -------------------------------- pkg/sfu/receiver.go | 5 ++--- 10 files changed, 82 insertions(+), 77 deletions(-) delete mode 100644 pkg/sfu/helpers_test.go diff --git a/go.mod b/go.mod index 9e63bdb05..54fd9c04c 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index b0d9308f5..869d1729d 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/pkg/rtc/datatrack.go b/pkg/rtc/datatrack.go index 6646abc95..81a619fed 100644 --- a/pkg/rtc/datatrack.go +++ b/pkg/rtc/datatrack.go @@ -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 } diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 74497c134..d5299b872 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -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() diff --git a/pkg/sfu/buffer/helpers.go b/pkg/sfu/buffer/helpers.go index 1f14c568c..294352135 100644 --- a/pkg/sfu/buffer/helpers.go +++ b/pkg/sfu/buffer/helpers.go @@ -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) +} + +// ------------------------------------------ diff --git a/pkg/sfu/buffer/helpers_test.go b/pkg/sfu/buffer/helpers_test.go index 219e65175..70b6a9935 100644 --- a/pkg/sfu/buffer/helpers_test.go +++ b/pkg/sfu/buffer/helpers_test.go @@ -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) + } + }) + } +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 7f4ac49a5..eb30545c7 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -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{ diff --git a/pkg/sfu/helpers.go b/pkg/sfu/helpers.go index fde31c33c..0003d5420 100644 --- a/pkg/sfu/helpers.go +++ b/pkg/sfu/helpers.go @@ -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)) } diff --git a/pkg/sfu/helpers_test.go b/pkg/sfu/helpers_test.go deleted file mode 100644 index f21bc84ac..000000000 --- a/pkg/sfu/helpers_test.go +++ /dev/null @@ -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) - } - }) - } -} diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index fe9d81738..e783307bd 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -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 {