From 5a426d15e1290391ebbcb3885795e44d75914b71 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 22 Oct 2025 14:09:33 +0530 Subject: [PATCH] Use rtp converter from protocol/utils (#4019) --- go.mod | 18 ++++++------- go.sum | 36 +++++++++++++------------- pkg/service/wire_gen.go | 14 +++++----- pkg/sfu/downtrack.go | 8 ++++++ pkg/sfu/rtpstats/rtpstats_base.go | 36 ++++++++++++++------------ pkg/sfu/rtpstats/rtpstats_base_lite.go | 8 +++--- pkg/sfu/rtpstats/rtpstats_receiver.go | 15 +++++------ pkg/sfu/rtpstats/rtpstats_sender.go | 13 +++++----- 8 files changed, 77 insertions(+), 71 deletions(-) diff --git a/go.mod b/go.mod index cf27ee43a..9eebfe200 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397 - github.com/livekit/protocol v1.42.2-0.20251015173118-0236ffc17f5e + github.com/livekit/protocol v1.42.3-0.20251022070801-a1e0e55a5c55 github.com/livekit/psrpc v0.7.0 github.com/mackerelio/go-osstat v0.2.6 github.com/magefile/mage v1.15.0 @@ -35,8 +35,8 @@ require ( github.com/pion/dtls/v3 v3.0.7 github.com/pion/ice/v4 v4.0.10 github.com/pion/interceptor v0.1.41 - github.com/pion/rtcp v1.2.15 - github.com/pion/rtp v1.8.23 + github.com/pion/rtcp v1.2.16 + github.com/pion/rtp v1.8.24 github.com/pion/sctp v1.8.40 github.com/pion/sdp/v3 v3.0.16 github.com/pion/transport/v3 v3.0.8 @@ -44,7 +44,7 @@ require ( github.com/pion/webrtc/v4 v4.1.6 github.com/pkg/errors v0.9.1 github.com/prometheus/client_golang v1.23.0 - github.com/redis/go-redis/v9 v9.14.0 + github.com/redis/go-redis/v9 v9.14.1 github.com/rs/cors v1.11.1 github.com/stretchr/testify v1.11.1 github.com/thoas/go-funk v0.9.3 @@ -55,7 +55,7 @@ require ( go.uber.org/atomic v1.11.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.27.0 - golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b + golang.org/x/exp v0.0.0-20251017212417-90e834f514db golang.org/x/mod v0.29.0 golang.org/x/sync v0.17.0 google.golang.org/protobuf v1.36.10 @@ -106,7 +106,7 @@ require ( github.com/hashicorp/go-retryablehttp v0.7.7 // indirect github.com/hashicorp/golang-lru v1.0.2 // indirect github.com/josharian/native v1.1.0 // indirect - github.com/klauspost/compress v1.18.0 // indirect + github.com/klauspost/compress v1.18.1 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/lithammer/shortuuid/v4 v4.2.0 // indirect github.com/mattn/go-runewidth v0.0.9 // indirect @@ -144,9 +144,9 @@ require ( golang.org/x/net v0.46.0 // indirect golang.org/x/sys v0.37.0 // indirect golang.org/x/text v0.30.0 // indirect - golang.org/x/tools v0.37.0 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20251014184007-4626949a642f // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20251014184007-4626949a642f // indirect + golang.org/x/tools v0.38.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20251020155222-88f65dc88635 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20251020155222-88f65dc88635 // indirect google.golang.org/grpc v1.76.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) diff --git a/go.sum b/go.sum index a1ebcfe48..69c0a5800 100644 --- a/go.sum +++ b/go.sum @@ -148,8 +148,8 @@ github.com/jxskiss/base62 v1.1.0 h1:A5zbF8v8WXx2xixnAKD2w+abC+sIzYJX+nxmhA6HWFw= github.com/jxskiss/base62 v1.1.0/go.mod h1:HhWAlUXvxKThfOlZbcuFzsqwtF5TcqS9ru3y5GfjWAc= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= -github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co= +github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= @@ -171,8 +171,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397 h1:Z7j2mY+bvG05UC80MpnJkitlJju8sSDWsr0Bb4dPceo= github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A= -github.com/livekit/protocol v1.42.2-0.20251015173118-0236ffc17f5e h1:2pgUHq8atL01StoCTrK760eYTKDoDz1nC4a90knEw5g= -github.com/livekit/protocol v1.42.2-0.20251015173118-0236ffc17f5e/go.mod h1:vhMS30QoEyH2p34vi6X1eWkC4EMV72ZGZwQb74ajY7A= +github.com/livekit/protocol v1.42.3-0.20251022070801-a1e0e55a5c55 h1:bWD8TWLbI+t+DIVUSyqKrxtTQHIkRQoY5Tkw5JpTwMQ= +github.com/livekit/protocol v1.42.3-0.20251022070801-a1e0e55a5c55/go.mod h1:m8IMgMd9FUi3HnCBQMDExVumbmVfiJy9/xKOifDAQtY= github.com/livekit/psrpc v0.7.0 h1:rtfqfjYN06WJYloE/S0nmkJ/Y04x4pxLQLe8kQ4FVHU= github.com/livekit/psrpc v0.7.0/go.mod h1:AuDC5uOoEjQJEc69v4Li3t77Ocz0e0NdjQEuFfO+vfk= github.com/mackerelio/go-osstat v0.2.6 h1:gs4U8BZeS1tjrL08tt5VUliVvSWP26Ai2Ob8Lr7f2i0= @@ -250,10 +250,10 @@ github.com/pion/mdns/v2 v2.0.7 h1:c9kM8ewCgjslaAmicYMFQIde2H9/lrZpjBkN8VwoVtM= github.com/pion/mdns/v2 v2.0.7/go.mod h1:vAdSYNAT0Jy3Ru0zl2YiW3Rm/fJCwIeM0nToenfOJKA= github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= -github.com/pion/rtcp v1.2.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo= -github.com/pion/rtcp v1.2.15/go.mod h1:jlGuAjHMEXwMUHK78RgX0UmEJFV4zUKOFHR7OP+D3D0= -github.com/pion/rtp v1.8.23 h1:kxX3bN4nM97DPrVBGq5I/Xcl332HnTHeP1Swx3/MCnU= -github.com/pion/rtp v1.8.23/go.mod h1:rF5nS1GqbR7H/TCpKwylzeq6yDM+MM6k+On5EgeThEM= +github.com/pion/rtcp v1.2.16 h1:fk1B1dNW4hsI78XUCljZJlC4kZOPk67mNRuQ0fcEkSo= +github.com/pion/rtcp v1.2.16/go.mod h1:/as7VKfYbs5NIb4h6muQ35kQF/J0ZVNz2Z3xKoCBYOo= +github.com/pion/rtp v1.8.24 h1:+ICyZXUQDv95EsHN70RrA4XKJf5MGWyC6QQc1u6/ynI= +github.com/pion/rtp v1.8.24/go.mod h1:rF5nS1GqbR7H/TCpKwylzeq6yDM+MM6k+On5EgeThEM= github.com/pion/sctp v1.8.40 h1:bqbgWYOrUhsYItEnRObUYZuzvOMsVplS3oNgzedBlG8= github.com/pion/sctp v1.8.40/go.mod h1:SPBBUENXE6ThkEksN5ZavfAhFYll+h+66ZiG6IZQuzo= github.com/pion/sdp/v3 v3.0.16 h1:0dKzYO6gTAvuLaAKQkC02eCPjMIi4NuAr/ibAwrGDCo= @@ -282,8 +282,8 @@ github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzM github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/puzpuzpuz/xsync/v3 v3.5.1 h1:GJYJZwO6IdxN/IKbneznS6yPkVC+c3zyY/j19c++5Fg= github.com/puzpuzpuz/xsync/v3 v3.5.1/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA= -github.com/redis/go-redis/v9 v9.14.0 h1:u4tNCjXOyzfgeLN+vAZaW1xUooqWDqVEsZN0U01jfAE= -github.com/redis/go-redis/v9 v9.14.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw= +github.com/redis/go-redis/v9 v9.14.1 h1:nDCrEiJmfOWhD76xlaw+HXT0c9hfNWeXgl0vIRYSDvQ= +github.com/redis/go-redis/v9 v9.14.1/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw= github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= @@ -369,8 +369,8 @@ golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1m golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04= golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0= -golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b h1:M2rDM6z3Fhozi9O7NWsxAkg/yqS/lQJ6PmkyIV3YP+o= -golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8= +golang.org/x/exp v0.0.0-20251017212417-90e834f514db h1:by6IehL4BH5k3e3SJmcoNbOobMey2SLpAF79iPOEBvw= +golang.org/x/exp v0.0.0-20251017212417-90e834f514db/go.mod h1:j/pmGrbnkbPtQfxEe5D0VQhZC6qKbfKifgD0oM7sR70= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= @@ -473,18 +473,18 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58= golang.org/x/tools v0.17.0/go.mod h1:xsh6VxdV005rRVaS6SSAf9oiAqljS7UZUacMZ8Bnsps= -golang.org/x/tools v0.37.0 h1:DVSRzp7FwePZW356yEAChSdNcQo6Nsp+fex1SUW09lE= -golang.org/x/tools v0.37.0/go.mod h1:MBN5QPQtLMHVdvsbtarmTNukZDdgwdwlO5qGacAzF0w= +golang.org/x/tools v0.38.0 h1:Hx2Xv8hISq8Lm16jvBZ2VQf+RLmbd7wVUsALibYI/IQ= +golang.org/x/tools v0.38.0/go.mod h1:yEsQ/d/YK8cjh0L6rZlY8tgtlKiBNTL14pGDJPJpYQs= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= -google.golang.org/genproto/googleapis/api v0.0.0-20251014184007-4626949a642f h1:OiFuztEyBivVKDvguQJYWq1yDcfAHIID/FVrPR4oiI0= -google.golang.org/genproto/googleapis/api v0.0.0-20251014184007-4626949a642f/go.mod h1:kprOiu9Tr0JYyD6DORrc4Hfyk3RFXqkQ3ctHEum3ZbM= -google.golang.org/genproto/googleapis/rpc v0.0.0-20251014184007-4626949a642f h1:1FTH6cpXFsENbPR5Bu8NQddPSaUUE6NA2XdZdDSAJK4= -google.golang.org/genproto/googleapis/rpc v0.0.0-20251014184007-4626949a642f/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= +google.golang.org/genproto/googleapis/api v0.0.0-20251020155222-88f65dc88635 h1:1wvBeYv+A2zfEbxROscJl69OP0m74S8wGEO+Syat26o= +google.golang.org/genproto/googleapis/api v0.0.0-20251020155222-88f65dc88635/go.mod h1:fDMmzKV90WSg1NbozdqrE64fkuTv6mlq2zxo9ad+3yo= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251020155222-88f65dc88635 h1:3uycTxukehWrxH4HtPRtn1PDABTU331ViDjyqrUbaog= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251020155222-88f65dc88635/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= google.golang.org/grpc v1.76.0 h1:UnVkv1+uMLYXoIz6o7chp59WfQUYA2ex/BXQ9rHZu7A= google.golang.org/grpc v1.76.0/go.mod h1:Ju12QI8M6iQJtbcsV+awF5a4hfJMLi4X0JLo94ULZ6c= google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 22b1e0c1c..b33cc1744 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -89,23 +89,23 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live } rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService, objectStore) topicFormatter := rpc.NewTopicFormatter() - v, err := rpc.NewTypedRoomClient(clientParams) + roomClient, err := rpc.NewTypedRoomClient(clientParams) if err != nil { return nil, err } - v2, err := rpc.NewTypedParticipantClient(clientParams) + participantClient, err := rpc.NewTypedParticipantClient(clientParams) if err != nil { return nil, err } - roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2) + roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient) if err != nil { return nil, err } - v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams) + agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams) if err != nil { return nil, err } - agentDispatchService := NewAgentDispatchService(v3, topicFormatter, roomAllocator, router) + agentDispatchService := NewAgentDispatchService(agentDispatchInternalClient, topicFormatter, roomAllocator, router) egressService := NewEgressService(egressClient, rtcEgressLauncher, ioInfoService, roomService) ingressConfig := getIngressConfig(conf) ingressClient, err := rpc.NewIngressClient(clientParams) @@ -120,11 +120,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live } sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService) rtcService := NewRTCService(conf, roomAllocator, router, telemetryService) - v4, err := rpc.NewTypedWHIPParticipantClient(clientParams) + whipParticipantClient, err := rpc.NewTypedWHIPParticipantClient(clientParams) if err != nil { return nil, err } - serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, v4) + serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, whipParticipantClient) if err != nil { return nil, err } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index d6ebd8d5b..a6556236a 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -847,6 +847,14 @@ func (d *DownTrack) setRTPHeaderExtensions() { d.absCaptureTimeExtID = ext.ID } } + d.params.Logger.Debugw( + "negotiated extension ids", + "absSendTimeExtID", d.absSendTimeExtID, + "dependencyDescriptorExtID", d.dependencyDescriptorExtID, + "playoutDelayExtID", d.playoutDelayExtID, + "transportWideExtID", d.transportWideExtID, + "absCaptureTimeExtID", d.absCaptureTimeExtID, + ) d.bindLock.Unlock() } diff --git a/pkg/sfu/rtpstats/rtpstats_base.go b/pkg/sfu/rtpstats/rtpstats_base.go index a4a9107ce..39e499ac1 100644 --- a/pkg/sfu/rtpstats/rtpstats_base.go +++ b/pkg/sfu/rtpstats/rtpstats_base.go @@ -205,6 +205,8 @@ func RTCPSenderReportPropagationDelay(rsrs *livekit.RTCPSenderReportState, passT type rtpStatsBase struct { *rtpStatsBaseLite + rtpConverter *utils.RTPConverter + firstTime int64 firstTimeAdjustment time.Duration highestTime int64 @@ -246,6 +248,7 @@ type rtpStatsBase struct { func newRTPStatsBase(params RTPStatsParams) *rtpStatsBase { return &rtpStatsBase{ rtpStatsBaseLite: newRTPStatsBaseLite(params), + rtpConverter: utils.NewRTPConverter(int64(params.ClockRate)), nextSnapshotID: cFirstSnapshotID, snapshots: make([]snapshot, 2), } @@ -385,7 +388,7 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime( srData *livekit.RTCPSenderReportState, tsOffset uint64, extStartTS uint64, -) (adjustment int64, err error, loggingFields []interface{}) { +) (adjustment int64, err error, loggingFields []any) { nowNano := mono.UnixNano() if time.Duration(nowNano-r.startTime) > cFirstPacketTimeAdjustWindow { return @@ -398,21 +401,21 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime( // in some network element along the way), push back first time // to an earlier instance. timeSinceReceive := time.Duration(nowNano - srData.AtAdjusted) - extNowTS := srData.RtpTimestampExt - tsOffset + uint64(timeSinceReceive.Nanoseconds()*int64(r.params.ClockRate)/1e9) + extNowTS := srData.RtpTimestampExt - tsOffset + r.rtpConverter.ToRTPExt(timeSinceReceive) samplesDiff := int64(extNowTS - extStartTS) if samplesDiff < 0 { // out-of-order, skip return } - samplesDuration := time.Duration(float64(samplesDiff) / float64(r.params.ClockRate) * float64(time.Second)) + samplesDuration := r.rtpConverter.ToDurationExt(uint64(samplesDiff)) timeSinceFirst := time.Duration(nowNano - r.firstTime) now := r.firstTime + timeSinceFirst.Nanoseconds() firstTime := now - samplesDuration.Nanoseconds() adjustment = r.firstTime - firstTime - getFields := func() []interface{} { - return []interface{}{ + getFields := func() []any { + return []any{ "startTime", time.Unix(0, r.startTime), "nowTime", time.Unix(0, now), "before", time.Unix(0, r.firstTime), @@ -461,7 +464,7 @@ func (r *rtpStatsBase) deltaInfo( snapshotID uint32, extStartSN uint64, extHighestSN uint64, -) (deltaInfo *RTPDeltaInfo, err error, loggingFields []interface{}) { +) (deltaInfo *RTPDeltaInfo, err error, loggingFields []any) { then, now := r.getAndResetSnapshot(snapshotID, extStartSN, extHighestSN) if now == nil || then == nil { return @@ -475,7 +478,7 @@ func (r *rtpStatsBase) deltaInfo( packetsExpected = 0 } if packetsExpected > cNumSequenceNumbers { - loggingFields = []interface{}{ + loggingFields = []any{ "snapshotID", snapshotID, "snapshotNow", now, "snapshotThen", then, @@ -501,7 +504,7 @@ func (r *rtpStatsBase) deltaInfo( // padding packets delta could be higher than expected due to out-of-order padding packets packetsPadding := now.packetsPadding - then.packetsPadding if packetsExpected < packetsPadding { - loggingFields = []interface{}{ + loggingFields = []any{ "snapshotID", snapshotID, "snapshotNow", now, "snapshotThen", then, @@ -651,7 +654,7 @@ func (r *rtpStatsBase) updateJitter(ets uint64, packetTime int64) float64 { // although it is the second packet of a frame because of out-of-order receival. if r.lastJitterExtTimestamp != ets { timeSinceFirst := packetTime - r.firstTime - packetTimeRTP := uint64(timeSinceFirst * int64(r.params.ClockRate) / 1e9) + packetTimeRTP := r.rtpConverter.ToRTPExt(time.Duration(timeSinceFirst)) transit := packetTimeRTP - ets if r.lastTransit != 0 { @@ -700,21 +703,20 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) ( rebasedReportDrift *livekit.RTPDrift, ) { if r.firstTime != 0 { - elapsed := r.highestTime - r.firstTime + elapsed := time.Duration(r.highestTime - r.firstTime) rtpClockTicks := extHighestTS - extStartTS - driftSamples := int64(rtpClockTicks - uint64(elapsed*int64(r.params.ClockRate)/1e9)) + driftSamples := int64(rtpClockTicks - r.rtpConverter.ToRTPExt(elapsed)) if elapsed > 0 { - elapsedSeconds := time.Duration(elapsed).Seconds() packetDrift = &livekit.RTPDrift{ StartTime: timestamppb.New(time.Unix(0, r.firstTime)), EndTime: timestamppb.New(time.Unix(0, r.highestTime)), - Duration: elapsedSeconds, + Duration: elapsed.Seconds(), StartTimestamp: extStartTS, EndTimestamp: extHighestTS, RtpClockTicks: rtpClockTicks, DriftSamples: driftSamples, DriftMs: (float64(driftSamples) * 1000) / float64(r.params.ClockRate), - ClockRate: float64(rtpClockTicks) / elapsedSeconds, + ClockRate: float64(rtpClockTicks) / elapsed.Seconds(), } } } @@ -724,7 +726,7 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) ( elapsed := mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time().Sub(mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time()) if elapsed.Seconds() > 0.0 { - driftSamples := int64(rtpClockTicks - uint64(elapsed.Nanoseconds()*int64(r.params.ClockRate)/1e9)) + driftSamples := int64(rtpClockTicks - r.rtpConverter.ToRTPExt(elapsed)) ntpReportDrift = &livekit.RTPDrift{ StartTime: timestamppb.New(mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time()), EndTime: timestamppb.New(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()), @@ -740,7 +742,7 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) ( elapsed = time.Duration(r.srNewest.At - r.srFirst.At) if elapsed.Seconds() > 0.0 { - driftSamples := int64(rtpClockTicks - uint64(elapsed.Nanoseconds()*int64(r.params.ClockRate)/1e9)) + driftSamples := int64(rtpClockTicks - r.rtpConverter.ToRTPExt(elapsed)) receivedReportDrift = &livekit.RTPDrift{ StartTime: timestamppb.New(time.Unix(0, r.srFirst.At)), EndTime: timestamppb.New(time.Unix(0, r.srNewest.At)), @@ -756,7 +758,7 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) ( elapsed = time.Duration(r.srNewest.AtAdjusted - r.srFirst.AtAdjusted) if elapsed.Seconds() > 0.0 { - driftSamples := int64(rtpClockTicks - uint64(elapsed.Nanoseconds()*int64(r.params.ClockRate)/1e9)) + driftSamples := int64(rtpClockTicks - r.rtpConverter.ToRTPExt(elapsed)) rebasedReportDrift = &livekit.RTPDrift{ StartTime: timestamppb.New(time.Unix(0, r.srFirst.AtAdjusted)), EndTime: timestamppb.New(time.Unix(0, r.srNewest.AtAdjusted)), diff --git a/pkg/sfu/rtpstats/rtpstats_base_lite.go b/pkg/sfu/rtpstats/rtpstats_base_lite.go index c54b3665d..026364c9a 100644 --- a/pkg/sfu/rtpstats/rtpstats_base_lite.go +++ b/pkg/sfu/rtpstats/rtpstats_base_lite.go @@ -319,7 +319,7 @@ func (r *rtpStatsBaseLite) deltaInfoLite( packetsExpected = 0 } if packetsExpected > cNumSequenceNumbers { - loggingFields = []interface{}{ + loggingFields = []any{ "snapshotLiteID", snapshotLiteID, "snapshotLiteNow", now, "snapshotLiteThen", then, @@ -342,7 +342,7 @@ func (r *rtpStatsBaseLite) deltaInfoLite( packetsLost = 0 } if packetsLost > packetsExpected { - loggingFields = []interface{}{ + loggingFields = []any{ "snapshotLiteID", snapshotLiteID, "snapshotLiteNow", now, "snapshotLiteThen", then, @@ -474,7 +474,7 @@ func (r *rtpStatsBaseLite) toProto(packetsExpected, packetsSeenMinusPadding, pac } gapsPresent := false - for i := 0; i < len(r.gapHistogram); i++ { + for i := range len(r.gapHistogram) { if r.gapHistogram[i] == 0 { continue } @@ -485,7 +485,7 @@ func (r *rtpStatsBaseLite) toProto(packetsExpected, packetsSeenMinusPadding, pac if gapsPresent { p.GapHistogram = make(map[int32]uint32, len(r.gapHistogram)) - for i := 0; i < len(r.gapHistogram); i++ { + for i := range len(r.gapHistogram) { if r.gapHistogram[i] == 0 { continue } diff --git a/pkg/sfu/rtpstats/rtpstats_receiver.go b/pkg/sfu/rtpstats/rtpstats_receiver.go index 4b2424f1c..a4a29dc08 100644 --- a/pkg/sfu/rtpstats/rtpstats_receiver.go +++ b/pkg/sfu/rtpstats/rtpstats_receiver.go @@ -116,10 +116,7 @@ func (r *RTPStatsReceiver) getTSRolloverCount(diffNano int64, ts uint32) int { } excess := (diffNano - r.tsRolloverThreshold*2) * int64(r.params.ClockRate) / 1e9 - roc := excess / (1 << 32) - if roc < 0 { - roc = 0 - } + roc := max(excess/(1<<32), 0) if r.timestamp.GetHighest() > ts { roc++ } @@ -230,7 +227,7 @@ func (r *RTPStatsReceiver) Update( // to before mute, but it appears like it has rolled over. // Use a threshold against expected to ignore these. if gapSN < 0 && gapTS > 0 { - expectedTSJump = timeSinceHighest * int64(r.params.ClockRate) / 1e9 + expectedTSJump = int64(r.rtpConverter.ToRTPExt(time.Duration(timeSinceHighest))) if gapTS > int64(float64(expectedTSJump)*cTSJumpTooHighFactor) { r.sequenceNumber.UndoUpdate(resSN) r.timestamp.UndoUpdate(resTS) @@ -364,7 +361,7 @@ func (r *RTPStatsReceiver) getExtendedSenderReport(srData *livekit.RTCPSenderRep // use time since last sender report to ensure long gaps where the time stamp might // jump more than half the range timeSinceLastReport := mediatransportutil.NtpTime(srData.NtpTimestamp).Time().Sub(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()) - expectedRTPTimestampExt := r.srNewest.RtpTimestampExt + uint64(timeSinceLastReport.Nanoseconds()*int64(r.params.ClockRate)/1e9) + expectedRTPTimestampExt := r.srNewest.RtpTimestampExt + r.rtpConverter.ToRTPExt(timeSinceLastReport) lbound := expectedRTPTimestampExt - uint64(cReportSlack*float64(r.params.ClockRate)) ubound := expectedRTPTimestampExt + uint64(cReportSlack*float64(r.params.ClockRate)) isInRange := (srData.RtpTimestamp-uint32(lbound) < (1 << 31)) && (uint32(ubound)-srData.RtpTimestamp < (1 << 31)) @@ -463,14 +460,14 @@ func (r *RTPStatsReceiver) checkRTPClockSkewAgainstMediaPathForSenderReport(srDa nowNano := mono.UnixNano() timeSinceSR := time.Duration(nowNano - srData.AtAdjusted) - extNowTSSR := srData.RtpTimestampExt + uint64(timeSinceSR.Nanoseconds()*int64(r.params.ClockRate)/1e9) + extNowTSSR := srData.RtpTimestampExt + r.rtpConverter.ToRTPExt(timeSinceSR) timeSinceHighest := time.Duration(nowNano - r.highestTime) - extNowTSHighest := r.timestamp.GetExtendedHighest() + uint64(timeSinceHighest.Nanoseconds()*int64(r.params.ClockRate)/1e9) + extNowTSHighest := r.timestamp.GetExtendedHighest() + r.rtpConverter.ToRTPExt(timeSinceHighest) diffHighest := extNowTSSR - extNowTSHighest timeSinceFirst := time.Duration(nowNano - r.firstTime) - extNowTSFirst := r.timestamp.GetExtendedStart() + uint64(timeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) + extNowTSFirst := r.timestamp.GetExtendedStart() + r.rtpConverter.ToRTPExt(timeSinceFirst) diffFirst := extNowTSSR - extNowTSFirst // is it more than 5 seconds off? diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index db43fa21c..7447dccac 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -815,8 +815,7 @@ func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt ui } timeDiff := at.Sub(time.Unix(0, r.firstTime)) - expectedRTPDiff := timeDiff.Nanoseconds() * int64(r.params.ClockRate) / 1e9 - expectedTSExt = r.extStartTS + uint64(expectedRTPDiff) + expectedTSExt = r.extStartTS + r.rtpConverter.ToRTPExt(timeDiff) return } @@ -834,20 +833,21 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek nowNTP mediatransportutil.NtpTime nowRTPExt uint64 ) + nowNano := mono.UnixNano() if passThrough { - timeSincePublisherSR := time.Duration(mono.UnixNano() - publisherSRData.At) + timeSincePublisherSR := time.Duration(nowNano - publisherSRData.At) reportTime = publisherSRData.At + timeSincePublisherSR.Nanoseconds() reportTimeAdjusted = publisherSRData.AtAdjusted + timeSincePublisherSR.Nanoseconds() nowNTP = mediatransportutil.ToNtpTime(mediatransportutil.NtpTime(publisherSRData.NtpTimestamp).Time().Add(timeSincePublisherSR)) - nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset + uint64(timeSincePublisherSR.Nanoseconds()*int64(r.params.ClockRate)/1e9) + nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset + r.rtpConverter.ToRTPExt(timeSincePublisherSR) } else { - timeSincePublisherSRAdjusted := time.Duration(mono.UnixNano() - publisherSRData.AtAdjusted) + timeSincePublisherSRAdjusted := time.Duration(nowNano - publisherSRData.AtAdjusted) reportTimeAdjusted = publisherSRData.AtAdjusted + timeSincePublisherSRAdjusted.Nanoseconds() reportTime = reportTimeAdjusted nowNTP = mediatransportutil.ToNtpTime(time.Unix(0, reportTime)) - nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset + uint64(timeSincePublisherSRAdjusted.Nanoseconds()*int64(r.params.ClockRate)/1e9) + nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset + r.rtpConverter.ToRTPExt(timeSincePublisherSRAdjusted) } packetCount := uint32(r.getPacketsSeenPlusDuplicates(r.extStartSN, r.extHighestSN)) @@ -863,7 +863,6 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek } ulgr := func() logger.UnlikelyLogger { - nowNano := mono.UnixNano() return r.logger.WithUnlikelyValues( "curr", WrappedRTCPSenderReportStateLogger{srData}, "feed", WrappedRTCPSenderReportStateLogger{publisherSRData},