diff --git a/CHANGELOG b/CHANGELOG index 68d7fb9ef45..1be930e8106 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -2,6 +2,67 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.8.3] - 2025-01-07 + +### Added + +- Allow requesting a dialtone during call transfer (#3122) +- Handle room configuration that's set in the grant itself (#3120) +- Update ICE to pick up accepting use-candidate unconditionally for ICE lite agents (#3150) +- auto create rooms during create agent dispatch api request (#3158) +- Annotate SIP errors with Twirp codes. (#3161) +- TWCC based congestion control (#3165 #3234 #3235 #3244 #3245 #3250 #3251 #3253 #3254 #3256 #3262 #3282) +- Loss based congestion signal detector. (#3168 #3169) +- Fix header size calculation in stats. (#3171) +- add per message deflate to signal ws (#3174) +- Add ResyncDownTracks API that can be used to resync all down tracks on (#3185) +- One shot signalling mode (#3188 #3192 #3194 #3223) +- Server side metrics (#3198) +- Add datastream packet type handling (#3210) +- Support SIP list filters. (#3240) +- Add RTX to downstream (#3247) +- Handle REMB on RTX RTCP (#3257) +- Thottle the publisher data channel sending when subscriber is slow (#3255 #3265 #3281) + +### Fixed + +- avoids NaN (#3119) +- reduce retransmit by seeding duplicate packets and bytes. (#3124) +- don't return video/rtx to client (#3142) +- ignore unexported fields in yaml lint (#3145) +- Fix incorrect computation of SecondsSinceNodeStatsUpdate (#3172) +- Attempt to fix missing participant left webhook. (#3173) +- Set down track connected flag in one-shot-signalling mode. (#3191) +- Don't SetCodecPreferences for video transceiver (#3249) +- Disable av1 for safari (#3284) +- fix completed job status updates causing workers to reconnect (#3294) + +### Changed + +- Display both pairs on selected candidate pair change (#3133) +- Maintain RTT marker for calculations. (#3139) +- Consolidate operations on LocalNode. (#3140) +- Use int64 nanoseconds and reduce conversion in a few places (#3159) +- De-centralize some configs to where they are used. (#3162) +- Split out audio level config. (#3163) +- Use int64 nanoseconds and reduce conversion in a few places (#3159) +- Reduce lock scope. (#3167) +- Clean up forwardRTP function a bit. (#3177) +- StreamAllocator (congestion controller) refactor (#3180) +- convert psprc error to http code in rtc service failure response (#3187) +- skip http request logging when the client aborts the request (#3195) +- Do not treat data publisher as publisher. (#3204) +- Publish data and signal bytes once every 30 seconds. (#3212) +- upgrade to pion/webrtc v4 (#3213) +- Don't wait rtp packet to fire track (#3246) +- Keep negotiated codec parameters in Downtrack.Bind (#3271) +- Structured logging of ParticipantInit (#3279) +- Start stream allocator after creating peer connection. (#3283) +- Reduce memory allocation in WritePaddingRTP / WriteProbePackets (#3288) +- add room/participant to logger context for SIP APIs (#3290) +- vp8 temporal layer selection with dependency descriptor (#3302) +- Use contiguous groups to determine queuing region. (#3308) + ## [1.8.0] - 2024-10-18 ### Added diff --git a/go.mod b/go.mod index ab65955731d..1cc2ba0e609 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/bep/debounce v1.2.1 github.com/d5/tengo/v2 v2.17.0 github.com/dustin/go-humanize v1.0.1 - github.com/elliotchance/orderedmap/v2 v2.6.1 + github.com/elliotchance/orderedmap/v2 v2.7.0 github.com/florianl/go-tc v0.4.4 github.com/frostbyte73/core v0.1.0 github.com/gammazero/deque v1.0.0 @@ -21,7 +21,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20241220010243-a2bdee945564 - github.com/livekit/protocol v1.30.0 + github.com/livekit/protocol v1.31.0 github.com/livekit/psrpc v0.6.1-0.20241018124827-1efff3d113a8 github.com/mackerelio/go-osstat v0.2.5 github.com/magefile/mage v1.15.0 @@ -31,15 +31,15 @@ require ( github.com/ory/dockertest/v3 v3.11.0 github.com/pion/datachannel v1.5.10 github.com/pion/dtls/v3 v3.0.4 - github.com/pion/ice/v4 v4.0.3 + github.com/pion/ice/v4 v4.0.5 github.com/pion/interceptor v0.1.37 github.com/pion/rtcp v1.2.15 - github.com/pion/rtp v1.8.9 + github.com/pion/rtp v1.8.11 github.com/pion/sctp v1.8.35 - github.com/pion/sdp/v3 v3.0.9 + github.com/pion/sdp/v3 v3.0.10 github.com/pion/transport/v3 v3.0.7 github.com/pion/turn/v4 v4.0.0 - github.com/pion/webrtc/v4 v4.0.6 + github.com/pion/webrtc/v4 v4.0.7 github.com/pkg/errors v0.9.1 github.com/prometheus/client_golang v1.20.5 github.com/redis/go-redis/v9 v9.7.0 @@ -53,9 +53,9 @@ 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-20241217172543-b2144cdd0a67 + golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8 golang.org/x/sync v0.10.0 - google.golang.org/protobuf v1.36.1 + google.golang.org/protobuf v1.36.3 gopkg.in/yaml.v3 v3.0.1 ) @@ -129,14 +129,14 @@ require ( github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect github.com/zeebo/xxh3 v1.0.2 // indirect go.uber.org/zap/exp v0.3.0 // indirect - golang.org/x/crypto v0.31.0 // indirect + golang.org/x/crypto v0.32.0 // indirect golang.org/x/mod v0.22.0 // indirect - golang.org/x/net v0.32.0 // indirect - golang.org/x/sys v0.28.0 // indirect + golang.org/x/net v0.34.0 // indirect + golang.org/x/sys v0.29.0 // indirect golang.org/x/text v0.21.0 // indirect - golang.org/x/tools v0.28.0 // indirect + golang.org/x/tools v0.29.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20241015192408-796eee8c2d53 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20241219192143-6b3ec007d9bb // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20250106144421-5f5ef82da422 // indirect google.golang.org/grpc v1.69.2 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) diff --git a/go.sum b/go.sum index 0749dd89ed6..0439fe2f72c 100644 --- a/go.sum +++ b/go.sum @@ -58,8 +58,8 @@ github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= -github.com/elliotchance/orderedmap/v2 v2.6.1 h1:vBgfLjyW0Nk0/NUgs/xiQoB38SVHATHcLTwxq2rrlC0= -github.com/elliotchance/orderedmap/v2 v2.6.1/go.mod h1:85lZyVbpGaGvHvnKa7Qhx7zncAdBIBq6u56Hb1PRU5Q= +github.com/elliotchance/orderedmap/v2 v2.7.0 h1:WHuf0DRo63uLnldCPp9ojm3gskYwEdIIfAUVG5KhoOc= +github.com/elliotchance/orderedmap/v2 v2.7.0/go.mod h1:85lZyVbpGaGvHvnKa7Qhx7zncAdBIBq6u56Hb1PRU5Q= github.com/envoyproxy/protoc-gen-validate v1.1.0 h1:tntQDh69XqOCOZsDz0lVJQez/2L6Uu2PdjCQwWCJ3bM= github.com/envoyproxy/protoc-gen-validate v1.1.0/go.mod h1:sXRDRVmzEbkM7CVcM06s9shE/m23dg3wzjl0UWqJ2q4= github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM= @@ -167,8 +167,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20241220010243-a2bdee945564 h1:GX7KF/V9ExmcfT/2Bdia8aROjkxrgx7WpyH7w9MB4J4= github.com/livekit/mediatransportutil v0.0.0-20241220010243-a2bdee945564/go.mod h1:36s+wwmU3O40IAhE+MjBWP3W71QRiEE9SfooSBvtBqY= -github.com/livekit/protocol v1.30.0 h1:yIaCrhRvfjWgAdvLkjf4VHh087yTeYrvjsNhLyXlUmE= -github.com/livekit/protocol v1.30.0/go.mod h1:08wT2rI6GecTCwh9n8OA28Gb7ZQNtIR+hX/LccP1TaY= +github.com/livekit/protocol v1.31.0 h1:o7u78rbPr+FWNJmiBfwqkA68RO/lie3JehWl3K/ejJQ= +github.com/livekit/protocol v1.31.0/go.mod h1:08wT2rI6GecTCwh9n8OA28Gb7ZQNtIR+hX/LccP1TaY= github.com/livekit/psrpc v0.6.1-0.20241018124827-1efff3d113a8 h1:Ibh0LoFl5NW5a1KFJEE0eLxxz7dqqKmYTj/BfCb0PbY= github.com/livekit/psrpc v0.6.1-0.20241018124827-1efff3d113a8/go.mod h1:CQUBSPfYYAaevg1TNCc6/aYsa8DJH4jSRFdCeSZk5u0= github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o= @@ -234,8 +234,8 @@ github.com/pion/datachannel v1.5.10 h1:ly0Q26K1i6ZkGf42W7D4hQYR90pZwzFOjTq5AuCKk github.com/pion/datachannel v1.5.10/go.mod h1:p/jJfC9arb29W7WrxyKbepTU20CFgyx5oLo8Rs4Py/M= github.com/pion/dtls/v3 v3.0.4 h1:44CZekewMzfrn9pmGrj5BNnTMDCFwr+6sLH+cCuLM7U= github.com/pion/dtls/v3 v3.0.4/go.mod h1:R373CsjxWqNPf6MEkfdy3aSe9niZvL/JaKlGeFphtMg= -github.com/pion/ice/v4 v4.0.3 h1:9s5rI1WKzF5DRqhJ+Id8bls/8PzM7mau0mj1WZb4IXE= -github.com/pion/ice/v4 v4.0.3/go.mod h1:VfHy0beAZ5loDT7BmJ2LtMtC4dbawIkkkejHPRZNB3Y= +github.com/pion/ice/v4 v4.0.5 h1:6awVfa1jg9YsI9/Lep4TG/o3kwS1Oayr5b8xz50ibJ8= +github.com/pion/ice/v4 v4.0.5/go.mod h1:JJaoEIxUIlGDA9gaRZbwXYqI3j6VG/QchpjX+QmwN6A= github.com/pion/interceptor v0.1.37 h1:aRA8Zpab/wE7/c0O3fh1PqY0AJI3fCSEM5lRWJVorwI= github.com/pion/interceptor v0.1.37/go.mod h1:JzxbJ4umVTlZAf+/utHzNesY8tmRkM2lVmkS82TTj8Y= github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY= @@ -246,12 +246,12 @@ 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.9 h1:E2HX740TZKaqdcPmf4pw6ZZuG8u5RlMMt+l3dxeu6Wk= -github.com/pion/rtp v1.8.9/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU= +github.com/pion/rtp v1.8.11 h1:17xjnY5WO5hgO6SD3/NTIUPvSFw/PbLsIJyz1r1yNIk= +github.com/pion/rtp v1.8.11/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4= github.com/pion/sctp v1.8.35 h1:qwtKvNK1Wc5tHMIYgTDJhfZk7vATGVHhXbUDfHbYwzA= github.com/pion/sctp v1.8.35/go.mod h1:EcXP8zCYVTRy3W9xtOF7wJm1L1aXfKRQzaM33SjQlzg= -github.com/pion/sdp/v3 v3.0.9 h1:pX++dCHoHUwq43kuwf3PyJfHlwIj4hXA7Vrifiq0IJY= -github.com/pion/sdp/v3 v3.0.9/go.mod h1:B5xmvENq5IXJimIO4zfp6LAe1fD9N+kFv+V/1lOdz8M= +github.com/pion/sdp/v3 v3.0.10 h1:6MChLE/1xYB+CjumMw+gZ9ufp2DPApuVSnDT8t5MIgA= +github.com/pion/sdp/v3 v3.0.10/go.mod h1:88GMahN5xnScv1hIMTqLdu/cOcUkj6a9ytbncwMCq2E= github.com/pion/srtp/v3 v3.0.4 h1:2Z6vDVxzrX3UHEgrUyIGM4rRouoC7v+NiF1IHtp9B5M= github.com/pion/srtp/v3 v3.0.4/go.mod h1:1Jx3FwDoxpRaTh1oRV8A/6G1BnFL+QI82eK4ms8EEJQ= github.com/pion/stun/v3 v3.0.0 h1:4h1gwhWLWuZWOJIJR9s2ferRO+W3zA/b6ijOI6mKzUw= @@ -260,8 +260,8 @@ github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1 github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo= github.com/pion/turn/v4 v4.0.0 h1:qxplo3Rxa9Yg1xXDxxH8xaqcyGUtbHYw4QSCvmFWvhM= github.com/pion/turn/v4 v4.0.0/go.mod h1:MuPDkm15nYSklKpN8vWJ9W2M0PlyQZqYt1McGuxG7mA= -github.com/pion/webrtc/v4 v4.0.6 h1:OfxfGeZGhneUDnZEoebLGDkzwjowSJ0avbOu2xaIUeM= -github.com/pion/webrtc/v4 v4.0.6/go.mod h1:j7oMHYvjl7lESJ/nYiE4d2URyjFbAo3uqJ6Xse6hbSg= +github.com/pion/webrtc/v4 v4.0.7 h1:aeq78uVnFZd2umXW0O9A2VFQYuS7+BZxWetQvSp2jPo= +github.com/pion/webrtc/v4 v4.0.7/go.mod h1:oFVBBVSHU3vAEwSgnk3BuKCwAUwpDwQhko1EDwyZWbU= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -294,15 +294,12 @@ github.com/stoewer/go-strcase v1.3.0/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8w github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= -github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= -github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= -github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/thoas/go-funk v0.9.3 h1:7+nAEx3kn5ZJcnDm2Bh23N2yOtweO14bi//dvRtgLpw= @@ -360,10 +357,10 @@ golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5y golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc= golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg= golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= -golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U= -golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= -golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67 h1:1UoZQm6f0P/ZO0w1Ri+f+ifG/gXhegadRdwBIXEFWDo= -golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67/go.mod h1:qj5a5QZpwLU2NLQudwIN5koi3beDhSAlJwa67PuM98c= +golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc= +golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc= +golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8 h1:yqrTHse8TCMW1M1ZCP+VAR/l0kKxwaAIqN/il7x4voA= +golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8/go.mod h1:tujkw807nyEEAamNbDrEGzRav+ilXA7PCRAd6xsmwiU= 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= @@ -394,8 +391,8 @@ golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk= golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY= -golang.org/x/net v0.32.0 h1:ZqPmj8Kzc+Y6e0+skZsuACbx+wzMgo5MQsJh9Qd6aYI= -golang.org/x/net v0.32.0/go.mod h1:CwU0IoeOlnQQWJ6ioyFrfRuomB8GKF6KbYXZVyeXNfs= +golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0= +golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -439,8 +436,8 @@ golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA= -golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU= +golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= @@ -466,20 +463,20 @@ 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.28.0 h1:WuB6qZ4RPCQo5aP3WdKZS7i595EdWqWR8vqJTlwTVK8= -golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw= +golang.org/x/tools v0.29.0 h1:Xx0h3TtM9rzQpQuR4dKLrdglAmCEN5Oi+P74JdhdzXE= +golang.org/x/tools v0.29.0/go.mod h1:KMQVMRsVxU6nHCFXrBPhDB8XncLNLM0lIy/F14RP588= 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= google.golang.org/genproto/googleapis/api v0.0.0-20241015192408-796eee8c2d53 h1:fVoAXEKA4+yufmbdVYv+SE73+cPZbbbe8paLsHfkK+U= google.golang.org/genproto/googleapis/api v0.0.0-20241015192408-796eee8c2d53/go.mod h1:riSXTwQ4+nqmPGtobMFyW5FqVAmIs0St6VPp4Ug7CE4= -google.golang.org/genproto/googleapis/rpc v0.0.0-20241219192143-6b3ec007d9bb h1:3oy2tynMOP1QbTC0MsNNAV+Se8M2Bd0A5+x1QHyw+pI= -google.golang.org/genproto/googleapis/rpc v0.0.0-20241219192143-6b3ec007d9bb/go.mod h1:lcTa1sDdWEIHMWlITnIczmw5w60CF9ffkb8Z+DVmmjA= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250106144421-5f5ef82da422 h1:3UsHvIr4Wc2aW4brOaSCmcxh9ksica6fHEr8P1XhkYw= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250106144421-5f5ef82da422/go.mod h1:3ENsm/5D1mzDyhpzeRi1NR784I0BcofWBoSc5QqqMK4= google.golang.org/grpc v1.69.2 h1:U3S9QEtbXC0bYNvRtcoklF3xGtLViumSYxWykJS+7AU= google.golang.org/grpc v1.69.2/go.mod h1:vyjdE6jLBI76dgpDojsFGNaHlxdjXN9ghpnd2o7JGZ4= -google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/gk= -google.golang.org/protobuf v1.36.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= +google.golang.org/protobuf v1.36.3 h1:82DV7MYdb8anAVi3qge1wSnMDrnKK7ebr+I0hHRN1BU= +google.golang.org/protobuf v1.36.3/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= diff --git a/pkg/rtc/clientinfo.go b/pkg/rtc/clientinfo.go index bd14391deed..2785edd0879 100644 --- a/pkg/rtc/clientinfo.go +++ b/pkg/rtc/clientinfo.go @@ -102,7 +102,8 @@ func (c ClientInfo) SupportErrorResponse() bool { } func (c ClientInfo) SupportSctpZeroChecksum() bool { - return !(c.isGo() && c.compareVersion("2.1.3") <= 0) + return !(c.ClientInfo.GetSdk() == livekit.ClientInfo_UNKNOWN || + (c.isGo() && c.compareVersion("2.4.0") < 0)) } // compareVersion compares a semver against the current client SDK version diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index 94f982d9884..1a87a0f9418 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -69,8 +69,6 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { // we don't want to use active TCP on a server, clients should be dialing webRTCConfig.SettingEngine.DisableActiveTCP(true) - webRTCConfig.SettingEngine.EnableSCTPZeroChecksum(true) - if rtcConf.PacketBufferSize == 0 { rtcConf.PacketBufferSize = 500 } @@ -120,12 +118,12 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { PacketBufferSizeAudio: rtcConf.PacketBufferSizeAudio, }, Publisher: publisherConfig, - Subscriber: getSubscriberConfig(rtcConf.CongestionControl), + Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE), }, nil } func (c *WebRTCConfig) UpdateCongestionControl(ccConf config.CongestionControlConfig) { - c.Subscriber = getSubscriberConfig(ccConf) + c.Subscriber = getSubscriberConfig(ccConf.UseSendSideBWEInterceptor || ccConf.UseSendSideBWE) } func (c *WebRTCConfig) SetBufferFactory(factory *buffer.Factory) { @@ -133,7 +131,7 @@ func (c *WebRTCConfig) SetBufferFactory(factory *buffer.Factory) { c.SettingEngine.BufferFactory = factory.GetOrNew } -func getSubscriberConfig(ccConf config.CongestionControlConfig) DirectionConfig { +func getSubscriberConfig(enableTWCC bool) DirectionConfig { subscriberConfig := DirectionConfig{ RTPHeaderExtension: RTPHeaderExtensionConfig{ Video: []string{ @@ -156,7 +154,7 @@ func getSubscriberConfig(ccConf config.CongestionControlConfig) DirectionConfig }, }, } - if ccConf.UseSendSideBWEInterceptor || ccConf.UseSendSideBWE { + if enableTWCC { subscriberConfig.RTPHeaderExtension.Video = append(subscriberConfig.RTPHeaderExtension.Video, sdp.TransportCCURI) subscriberConfig.RTCPFeedback.Video = append(subscriberConfig.RTCPFeedback.Video, webrtc.RTCPFeedback{Type: webrtc.TypeRTCPFBTransportCC}) } else { diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index fbcbdd2286c..725953fefc3 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -253,7 +253,7 @@ type ParticipantImpl struct { supervisor *supervisor.ParticipantSupervisor - tracksQuality map[livekit.TrackID]livekit.ConnectionQuality + connectionQuality livekit.ConnectionQuality metricTimestamper *metric.MetricTimestamper metricsCollector *metric.MetricsCollector @@ -292,9 +292,9 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { params.SID, params.Telemetry, ), - tracksQuality: make(map[livekit.TrackID]livekit.ConnectionQuality), - pubLogger: params.Logger.WithComponent(sutils.ComponentPub), - subLogger: params.Logger.WithComponent(sutils.ComponentSub), + connectionQuality: livekit.ConnectionQuality_EXCELLENT, + pubLogger: params.Logger.WithComponent(sutils.ComponentPub), + subLogger: params.Logger.WithComponent(sutils.ComponentSub), } if !params.DisableSupervisor { p.supervisor = supervisor.NewParticipantSupervisor(supervisor.ParticipantSupervisorParams{Logger: params.Logger}) @@ -1249,17 +1249,10 @@ func (p *ParticipantImpl) OnICEConfigChanged(f func(participant types.LocalParti } func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo { - numTracks := 0 minQuality := livekit.ConnectionQuality_EXCELLENT minScore := connectionquality.MaxMOS - numUpDrops := 0 - numDownDrops := 0 - - availableTracks := make(map[livekit.TrackID]bool) for _, pt := range p.GetPublishedTracks() { - numTracks++ - score, quality := pt.(types.LocalMediaTrack).GetConnectionScoreAndQuality() if utils.IsConnectionQualityLower(minQuality, quality) { minQuality = quality @@ -1267,24 +1260,10 @@ func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo } else if quality == minQuality && score < minScore { minScore = score } - - p.lock.Lock() - trackID := pt.ID() - if prevQuality, ok := p.tracksQuality[trackID]; ok { - if utils.IsConnectionQualityLower(prevQuality, quality) { - numUpDrops++ - } - } - p.tracksQuality[trackID] = quality - p.lock.Unlock() - - availableTracks[trackID] = true } subscribedTracks := p.SubscriptionManager.GetSubscribedTracks() for _, subTrack := range subscribedTracks { - numTracks++ - score, quality := subTrack.DownTrack().GetConnectionScoreAndQuality() if utils.IsConnectionQualityLower(minQuality, quality) { minQuality = quality @@ -1292,35 +1271,21 @@ func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo } else if quality == minQuality && score < minScore { minScore = score } + } - p.lock.Lock() - trackID := subTrack.ID() - if prevQuality, ok := p.tracksQuality[trackID]; ok { - if utils.IsConnectionQualityLower(prevQuality, quality) { - numDownDrops++ - } - } - p.tracksQuality[trackID] = quality - p.lock.Unlock() + prometheus.RecordQuality(minQuality, minScore) - availableTracks[trackID] = true + if minQuality == livekit.ConnectionQuality_LOST && !p.ProtocolVersion().SupportsConnectionQualityLost() { + minQuality = livekit.ConnectionQuality_POOR } - prometheus.RecordQuality(minQuality, minScore, numUpDrops, numDownDrops) - - // remove unavailable tracks from track quality cache p.lock.Lock() - for trackID := range p.tracksQuality { - if !availableTracks[trackID] { - delete(p.tracksQuality, trackID) - } + if minQuality != p.connectionQuality { + p.params.Logger.Debugw("connection quality changed", "from", p.connectionQuality, "to", minQuality) } + p.connectionQuality = minQuality p.lock.Unlock() - if minQuality == livekit.ConnectionQuality_LOST && !p.ProtocolVersion().SupportsConnectionQualityLost() { - minQuality = livekit.ConnectionQuality_POOR - } - return &livekit.ConnectionQualityInfo{ ParticipantSid: string(p.ID()), Quality: minQuality, @@ -1945,6 +1910,10 @@ func (p *ParticipantImpl) onDataMessage(kind livekit.DataPacket_Kind, data []byt if payload.StreamChunk == nil { return } + case *livekit.DataPacket_StreamTrailer: + if payload.StreamTrailer == nil { + return + } default: p.pubLogger.Warnw("received unsupported data packet", nil, "payload", payload) } diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 9923cab9c87..2e0a1081c99 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -332,12 +332,17 @@ func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error { err := sink.WriteMessage(msg) if errors.Is(err, psrpc.Canceled) { - p.params.Logger.Debugw("could not send message to participant", - "error", err, "messageType", fmt.Sprintf("%T", msg.Message)) + p.params.Logger.Debugw( + "could not send message to participant", + "error", err, + "messageType", fmt.Sprintf("%T", msg.Message), + ) return nil } else if err != nil { - p.params.Logger.Warnw("could not send message to participant", err, - "messageType", fmt.Sprintf("%T", msg.Message)) + p.params.Logger.Warnw( + "could not send message to participant", err, + "messageType", fmt.Sprintf("%T", msg.Message), + ) return err } return nil diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 412f787b158..6cb8ee129b3 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -830,7 +830,7 @@ func (r *Room) UpdateSubscriptionPermission(participant types.LocalParticipant, return nil } -func (r *Room) ResolveMediaTrackForSubscriber(subIdentity livekit.ParticipantIdentity, trackID livekit.TrackID) types.MediaResolverResult { +func (r *Room) ResolveMediaTrackForSubscriber(sub types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult { res := types.MediaResolverResult{} info := r.trackManager.GetTrackInfo(trackID) @@ -848,7 +848,7 @@ func (r *Room) ResolveMediaTrackForSubscriber(subIdentity livekit.ParticipantIde pub := r.GetParticipantByID(info.PublisherID) // when publisher is not found, we will assume it doesn't have permission to access if pub != nil { - res.HasPermission = pub.HasPermission(trackID, subIdentity) + res.HasPermission = IsParticipantExemptFromTrackPermissionsRestrictions(sub) || pub.HasPermission(trackID, sub.Identity()) } return res @@ -1809,6 +1809,12 @@ func IsCloseNotifySkippable(closeReason types.ParticipantCloseReason) bool { return closeReason == types.ParticipantCloseReasonDuplicateIdentity } +func IsParticipantExemptFromTrackPermissionsRestrictions(p types.LocalParticipant) bool { + // egress/recorder participants bypass permissions as auto-egress does not + // have enough context to check permissions + return p.IsRecorder() +} + func connectionDetailsFields(infos []*types.ICEConnectionInfo) []interface{} { var fields []interface{} connectionType := types.ICEConnectionTypeUnknown diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 927b14a1d1d..be97b2453df 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -503,7 +503,7 @@ func (m *SubscriptionManager) subscribe(s *trackSubscription) error { } trackID := s.trackID - res := m.params.TrackResolver(m.params.Participant.Identity(), trackID) + res := m.params.TrackResolver(m.params.Participant, trackID) s.logger.Debugw("resolved track", "result", res) if res.TrackChangedNotifier != nil && s.setChangedNotifier(res.TrackChangedNotifier) { @@ -517,7 +517,7 @@ func (m *SubscriptionManager) subscribe(s *trackSubscription) error { if res.TrackRemovedNotifier != nil && s.setRemovedNotifier(res.TrackRemovedNotifier) { res.TrackRemovedNotifier.AddObserver(string(m.params.Participant.ID()), func() { // re-resolve the track in case the same track had been re-published - res := m.params.TrackResolver(m.params.Participant.Identity(), trackID) + res := m.params.TrackResolver(m.params.Participant, trackID) if res.Track != nil { // do not unsubscribe, track is still available return @@ -627,7 +627,7 @@ func (m *SubscriptionManager) subscribeSynchronous(trackID livekit.TrackID) erro return ErrNoSubscribePermission } - res := m.params.TrackResolver(m.params.Participant.Identity(), trackID) + res := m.params.TrackResolver(m.params.Participant, trackID) m.params.Logger.Debugw("resolved track", "trackID", trackID, " result", res) track := res.Track diff --git a/pkg/rtc/subscriptionmanager_test.go b/pkg/rtc/subscriptionmanager_test.go index f44b9915329..a8dfe7032df 100644 --- a/pkg/rtc/subscriptionmanager_test.go +++ b/pkg/rtc/subscriptionmanager_test.go @@ -45,7 +45,7 @@ const ( func TestSubscribe(t *testing.T) { t.Run("happy path subscribe", func(t *testing.T) { - sm := newTestSubscriptionManager(t) + sm := newTestSubscriptionManager() defer sm.Close(false) resolver := newTestResolver(true, true, "pub", "pubID") sm.params.TrackResolver = resolver.Resolve @@ -118,7 +118,7 @@ func TestSubscribe(t *testing.T) { }) t.Run("no track permission", func(t *testing.T) { - sm := newTestSubscriptionManager(t) + sm := newTestSubscriptionManager() defer sm.Close(false) resolver := newTestResolver(false, true, "pub", "pubID") sm.params.TrackResolver = resolver.Resolve @@ -159,7 +159,7 @@ func TestSubscribe(t *testing.T) { }) t.Run("publisher left", func(t *testing.T) { - sm := newTestSubscriptionManager(t) + sm := newTestSubscriptionManager() defer sm.Close(false) resolver := newTestResolver(true, true, "pub", "pubID") sm.params.TrackResolver = resolver.Resolve @@ -188,7 +188,7 @@ func TestSubscribe(t *testing.T) { } func TestUnsubscribe(t *testing.T) { - sm := newTestSubscriptionManager(t) + sm := newTestSubscriptionManager() defer sm.Close(false) unsubCount := atomic.Int32{} sm.params.OnTrackUnsubscribed = func(subTrack types.SubscribedTrack) { @@ -208,7 +208,7 @@ func TestUnsubscribe(t *testing.T) { logger: logger.GetLogger(), } // a bunch of unfortunate manual wiring - res := resolver.Resolve("sub", s.trackID) + res := resolver.Resolve(nil, s.trackID) res.TrackChangedNotifier.AddObserver(string(sm.params.Participant.ID()), func() {}) s.changedNotifier = res.TrackChangedNotifier st, err := res.Track.AddSubscriber(sm.params.Participant) @@ -257,7 +257,7 @@ func TestUnsubscribe(t *testing.T) { } func TestSubscribeStatusChanged(t *testing.T) { - sm := newTestSubscriptionManager(t) + sm := newTestSubscriptionManager() defer sm.Close(false) resolver := newTestResolver(true, true, "pub", "pubID") sm.params.TrackResolver = resolver.Resolve @@ -320,7 +320,7 @@ func TestSubscribeStatusChanged(t *testing.T) { // clients may send update subscribed settings prior to subscription events coming through // settings should be persisted and used when the subscription does take place. func TestUpdateSettingsBeforeSubscription(t *testing.T) { - sm := newTestSubscriptionManager(t) + sm := newTestSubscriptionManager() defer sm.Close(false) resolver := newTestResolver(true, true, "pub", "pubID") sm.params.TrackResolver = resolver.Resolve @@ -351,7 +351,7 @@ func TestUpdateSettingsBeforeSubscription(t *testing.T) { } func TestSubscriptionLimits(t *testing.T) { - sm := newTestSubscriptionManagerWithParams(t, testSubscriptionParams{ + sm := newTestSubscriptionManagerWithParams(testSubscriptionParams{ SubscriptionLimitAudio: 1, SubscriptionLimitVideo: 1, }) @@ -452,11 +452,11 @@ type testSubscriptionParams struct { SubscriptionLimitVideo int32 } -func newTestSubscriptionManager(t *testing.T) *SubscriptionManager { - return newTestSubscriptionManagerWithParams(t, testSubscriptionParams{}) +func newTestSubscriptionManager() *SubscriptionManager { + return newTestSubscriptionManagerWithParams(testSubscriptionParams{}) } -func newTestSubscriptionManagerWithParams(t *testing.T, params testSubscriptionParams) *SubscriptionManager { +func newTestSubscriptionManagerWithParams(params testSubscriptionParams) *SubscriptionManager { p := &typesfakes.FakeLocalParticipant{} p.CanSubscribeReturns(true) p.IDReturns("subID") @@ -468,7 +468,7 @@ func newTestSubscriptionManagerWithParams(t *testing.T, params testSubscriptionP OnTrackSubscribed: func(subTrack types.SubscribedTrack) {}, OnTrackUnsubscribed: func(subTrack types.SubscribedTrack) {}, OnSubscriptionError: func(trackID livekit.TrackID, fatal bool, err error) {}, - TrackResolver: func(identity livekit.ParticipantIdentity, trackID livekit.TrackID) types.MediaResolverResult { + TrackResolver: func(sub types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult { return types.MediaResolverResult{} }, Telemetry: &telemetryfakes.FakeTelemetryService{}, @@ -502,7 +502,7 @@ func (t *testResolver) SetPause(paused bool) { t.paused = paused } -func (t *testResolver) Resolve(identity livekit.ParticipantIdentity, trackID livekit.TrackID) types.MediaResolverResult { +func (t *testResolver) Resolve(_subscriber types.LocalParticipant, trackID livekit.TrackID) types.MediaResolverResult { t.lock.Lock() defer t.lock.Unlock() res := types.MediaResolverResult{ diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 8813194476b..a274548ecf2 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -306,6 +306,10 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat se.SetFireOnTrackBeforeFirstRTP(true) } + if params.ClientInfo.SupportSctpZeroChecksum() { + se.EnableSCTPZeroChecksum(true) + } + // // Disable SRTP replay protection (https://datatracker.ietf.org/doc/html/rfc3711#page-15). // Needed due to lack of RTX stream support in Pion. @@ -1415,7 +1419,7 @@ func (t *PCTransport) initPCWithPreviousAnswer(previousAnswer webrtc.SessionDesc func (t *PCTransport) SetPreviousSdp(offer, answer *webrtc.SessionDescription) { // when there is no previous answer, cannot migrate, force a full reconnect if answer == nil { - t.params.Handler.OnNegotiationFailed() + t.onNegotiationFailed(true, "no previous answer for previous sdp") return } @@ -1423,10 +1427,9 @@ func (t *PCTransport) SetPreviousSdp(offer, answer *webrtc.SessionDescription) { if t.pc.RemoteDescription() == nil && t.previousAnswer == nil { t.previousAnswer = answer if senders, err := t.initPCWithPreviousAnswer(*t.previousAnswer); err != nil { - t.params.Logger.Warnw("initPCWithPreviousAnswer failed", err) t.lock.Unlock() - t.params.Handler.OnNegotiationFailed() + t.onNegotiationFailed(true, fmt.Sprintf("initPCWithPreviousAnswer failed, error: %s", err)) return } else if offer != nil { // in migration case, can't reuse transceiver before negotiated except track subscribed at previous node @@ -1491,8 +1494,7 @@ func (t *PCTransport) postEvent(e event) { } if err != nil { if !e.isClosed.Load() { - e.params.Logger.Warnw("error handling event", err, "event", e.String()) - e.params.Handler.OnNegotiationFailed() + e.onNegotiationFailed(true, fmt.Sprintf("error handling event. err: %s, event: %s", err, e)) } } }, e) @@ -1698,14 +1700,7 @@ func (t *PCTransport) setupSignalStateCheckTimer() { failed := t.negotiationState != transport.NegotiationStateNone if t.negotiateCounter.Load() == negotiateVersion && failed && t.pc.ConnectionState() == webrtc.PeerConnectionStateConnected { - t.params.Logger.Infow( - "negotiation timed out", - "localCurrent", t.pc.CurrentLocalDescription(), - "localPending", t.pc.PendingLocalDescription(), - "remoteCurrent", t.pc.CurrentRemoteDescription(), - "remotePending", t.pc.PendingRemoteDescription(), - ) - t.params.Handler.OnNegotiationFailed() + t.onNegotiationFailed(false, "negotiation timed out") } }) } @@ -2062,6 +2057,26 @@ func (t *PCTransport) handleICERestart(_ event) error { return t.doICERestart() } +func (t *PCTransport) onNegotiationFailed(warning bool, reason string) { + logFields := []interface{}{ + "reason", reason, + "localCurrent", t.pc.CurrentLocalDescription(), + "localPending", t.pc.PendingLocalDescription(), + "remoteCurrent", t.pc.CurrentRemoteDescription(), + "remotePending", t.pc.PendingRemoteDescription(), + } + if warning { + t.params.Logger.Warnw( + "negotiation failed", + nil, + logFields..., + ) + } else { + t.params.Logger.Infow("negotiation failed", logFields...) + } + t.params.Handler.OnNegotiationFailed() +} + // configure subscriber transceiver for audio stereo and nack // pion doesn't support per transciver codec configuration, so the nack of this session will be disabled // forever once it is first disabled by a transceiver. diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index d8cbeb0ac1b..e5855698646 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -476,7 +476,7 @@ type Room interface { UpdateSubscriptionPermission(participant LocalParticipant, permissions *livekit.SubscriptionPermission) error SyncState(participant LocalParticipant, state *livekit.SyncState) error SimulateScenario(participant LocalParticipant, scenario *livekit.SimulateScenario) error - ResolveMediaTrackForSubscriber(subIdentity livekit.ParticipantIdentity, trackID livekit.TrackID) MediaResolverResult + ResolveMediaTrackForSubscriber(sub LocalParticipant, trackID livekit.TrackID) MediaResolverResult GetLocalParticipants() []LocalParticipant } @@ -593,7 +593,7 @@ type MediaResolverResult struct { } // MediaTrackResolver locates a specific media track for a subscriber -type MediaTrackResolver func(livekit.ParticipantIdentity, livekit.TrackID) MediaResolverResult +type MediaTrackResolver func(LocalParticipant, livekit.TrackID) MediaResolverResult // Supervisor/operation monitor related definitions type OperationMonitorEvent int diff --git a/pkg/rtc/types/typesfakes/fake_room.go b/pkg/rtc/types/typesfakes/fake_room.go index 576e8ddcfe2..0b0a0d3416d 100644 --- a/pkg/rtc/types/typesfakes/fake_room.go +++ b/pkg/rtc/types/typesfakes/fake_room.go @@ -46,10 +46,10 @@ type FakeRoom struct { arg2 livekit.ParticipantID arg3 types.ParticipantCloseReason } - ResolveMediaTrackForSubscriberStub func(livekit.ParticipantIdentity, livekit.TrackID) types.MediaResolverResult + ResolveMediaTrackForSubscriberStub func(types.LocalParticipant, livekit.TrackID) types.MediaResolverResult resolveMediaTrackForSubscriberMutex sync.RWMutex resolveMediaTrackForSubscriberArgsForCall []struct { - arg1 livekit.ParticipantIdentity + arg1 types.LocalParticipant arg2 livekit.TrackID } resolveMediaTrackForSubscriberReturns struct { @@ -299,11 +299,11 @@ func (fake *FakeRoom) RemoveParticipantArgsForCall(i int) (livekit.ParticipantId return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } -func (fake *FakeRoom) ResolveMediaTrackForSubscriber(arg1 livekit.ParticipantIdentity, arg2 livekit.TrackID) types.MediaResolverResult { +func (fake *FakeRoom) ResolveMediaTrackForSubscriber(arg1 types.LocalParticipant, arg2 livekit.TrackID) types.MediaResolverResult { fake.resolveMediaTrackForSubscriberMutex.Lock() ret, specificReturn := fake.resolveMediaTrackForSubscriberReturnsOnCall[len(fake.resolveMediaTrackForSubscriberArgsForCall)] fake.resolveMediaTrackForSubscriberArgsForCall = append(fake.resolveMediaTrackForSubscriberArgsForCall, struct { - arg1 livekit.ParticipantIdentity + arg1 types.LocalParticipant arg2 livekit.TrackID }{arg1, arg2}) stub := fake.ResolveMediaTrackForSubscriberStub @@ -325,13 +325,13 @@ func (fake *FakeRoom) ResolveMediaTrackForSubscriberCallCount() int { return len(fake.resolveMediaTrackForSubscriberArgsForCall) } -func (fake *FakeRoom) ResolveMediaTrackForSubscriberCalls(stub func(livekit.ParticipantIdentity, livekit.TrackID) types.MediaResolverResult) { +func (fake *FakeRoom) ResolveMediaTrackForSubscriberCalls(stub func(types.LocalParticipant, livekit.TrackID) types.MediaResolverResult) { fake.resolveMediaTrackForSubscriberMutex.Lock() defer fake.resolveMediaTrackForSubscriberMutex.Unlock() fake.ResolveMediaTrackForSubscriberStub = stub } -func (fake *FakeRoom) ResolveMediaTrackForSubscriberArgsForCall(i int) (livekit.ParticipantIdentity, livekit.TrackID) { +func (fake *FakeRoom) ResolveMediaTrackForSubscriberArgsForCall(i int) (types.LocalParticipant, livekit.TrackID) { fake.resolveMediaTrackForSubscriberMutex.RLock() defer fake.resolveMediaTrackForSubscriberMutex.RUnlock() argsForCall := fake.resolveMediaTrackForSubscriberArgsForCall[i] diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index fcee84634ab..40f2f51e527 100644 --- a/pkg/service/ingress.go +++ b/pkg/service/ingress.go @@ -159,6 +159,7 @@ func (s *IngressService) CreateIngressWithUrl(ctx context.Context, urlStr string ParticipantName: req.ParticipantName, ParticipantMetadata: req.ParticipantMetadata, State: &livekit.IngressState{}, + Enabled: req.Enabled, } switch req.InputType { @@ -261,6 +262,10 @@ func updateInfoUsingRequest(req *livekit.UpdateIngressRequest, info *livekit.Ing info.Video = req.Video } + if req.Enabled != nil { + info.Enabled = req.Enabled + } + if err := ingress.ValidateForSerialization(info); err != nil { return err } diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index d9038bbc8f2..53416878489 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -49,7 +49,6 @@ import ( ) const ( - roomPurgeSeconds = 24 * 60 * 60 tokenRefreshInterval = 5 * time.Minute tokenDefaultTTL = 10 * time.Minute ) @@ -192,26 +191,6 @@ func (r *RoomManager) deleteRoom(ctx context.Context, roomName livekit.RoomName) return err } -// CleanupRooms cleans up after old rooms that have been around for a while -func (r *RoomManager) CleanupRooms() error { - // cleanup rooms that have been left for over a day - ctx := context.Background() - rooms, err := r.roomStore.ListRooms(ctx, nil) - if err != nil { - return err - } - - now := time.Now().Unix() - for _, room := range rooms { - if (now - room.CreationTime) > roomPurgeSeconds { - if err := r.deleteRoom(ctx, livekit.RoomName(room.Name)); err != nil { - return err - } - } - } - return nil -} - func (r *RoomManager) CloseIdleRooms() { r.lock.RLock() rooms := maps.Values(r.rooms) diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index d9d52eda366..ca3af76aef2 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -344,9 +344,11 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { } res, ok := msg.(*livekit.SignalResponse) if !ok { - pLogger.Errorw("unexpected message type", nil, + pLogger.Errorw( + "unexpected message type", nil, "type", fmt.Sprintf("%T", msg), - "connID", cr.ConnectionID) + "connID", cr.ConnectionID, + ) continue } @@ -356,10 +358,14 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { case *livekit.SignalResponse_Answer: pLogger.Debugw("sending answer", "answer", m) case *livekit.SignalResponse_Join: + pLogger.Debugw("sending join", "join", m) signalStats.ResolveRoom(m.Join.GetRoom()) signalStats.ResolveParticipant(m.Join.GetParticipant()) case *livekit.SignalResponse_RoomUpdate: + pLogger.Debugw("sending room update", "roomUpdate", m) signalStats.ResolveRoom(m.RoomUpdate.GetRoom()) + case *livekit.SignalResponse_Update: + pLogger.Debugw("sending participant update", "participantUpdate", m) } if count, err := sigConn.WriteResponse(res); err != nil { diff --git a/pkg/service/server.go b/pkg/service/server.go index 20ae0a4bb48..d4e1603a92e 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -160,10 +160,6 @@ func NewLivekitServer(conf *config.Config, } } - // clean up old rooms on startup - if err = roomManager.CleanupRooms(); err != nil { - return - } if err = router.RemoveDeadNodes(); err != nil { return } diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 8e79e0d316b..bc106d13e75 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -229,7 +229,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili for _, ext := range params.HeaderExtensions { switch ext.URI { case dd.ExtensionURI: - if IsSvcCodec(codec.MimeType) { + if IsSvcCodec(codec.MimeType) || strings.EqualFold(codec.MimeType, webrtc.MimeTypeVP8) { if b.ddExtID != 0 { b.logger.Warnw("multiple dependency descriptor extensions found", nil, "id", ext.ID, "previous", b.ddExtID) continue @@ -785,8 +785,9 @@ func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64, flowStat // DD-TODO : notify active decode target change if changed. } } - switch b.mime { - case "video/vp8": + + switch utils.MatchMimeType(b.mime) { + case utils.MimeTypeVP8: vp8Packet := VP8{} if err := vp8Packet.Unmarshal(rtpPacket.Payload); err != nil { b.logger.Warnw("could not unmarshal VP8 packet", err) @@ -802,7 +803,7 @@ func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64, flowStat ep.Payload = vp8Packet ep.Spatial = InvalidLayerSpatial // vp8 don't have spatial scalability, reset to invalid - case "video/vp9": + case utils.MimeTypeVP9: if ep.DependencyDescriptor == nil { var vp9Packet codecs.VP9Packet _, err := vp9Packet.Unmarshal(rtpPacket.Payload) @@ -818,11 +819,11 @@ func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64, flowStat } ep.KeyFrame = IsVP9KeyFrame(rtpPacket.Payload) - case "video/h264": + case utils.MimeTypeH264: ep.KeyFrame = IsH264KeyFrame(rtpPacket.Payload) ep.Spatial = InvalidLayerSpatial // h.264 don't have spatial scalability, reset to invalid - case "video/av1": + case utils.MimeTypeAV1: ep.KeyFrame = IsAV1KeyFrame(rtpPacket.Payload) } @@ -1122,10 +1123,8 @@ func (b *Buffer) GetTemporalLayerFpsForSpatial(layer int32) []float32 { // SVC-TODO: May only need to differentiate between simulcast and non-simulcast // SVC-TODO: i. e. may be possible to treat single layer as SVC to get proper/intended functionality. func IsSvcCodec(mime string) bool { - switch strings.ToLower(mime) { - case "video/av1": - fallthrough - case "video/vp9": + switch utils.MatchMimeType(mime) { + case utils.MimeTypeAV1, utils.MimeTypeVP9: return true } return false diff --git a/pkg/sfu/buffer/buffer_test.go b/pkg/sfu/buffer/buffer_test.go index de8f3302234..ad8a553b8d0 100644 --- a/pkg/sfu/buffer/buffer_test.go +++ b/pkg/sfu/buffer/buffer_test.go @@ -322,5 +322,4 @@ func BenchmarkMemcpu(b *testing.B) { for i := 0; i < b.N; i++ { copy(buf2, buf) } - } diff --git a/pkg/sfu/bwe/sendsidebwe/congestion_detector.go b/pkg/sfu/bwe/sendsidebwe/congestion_detector.go index 13ca8ce9348..c7f4092a6e2 100644 --- a/pkg/sfu/bwe/sendsidebwe/congestion_detector.go +++ b/pkg/sfu/bwe/sendsidebwe/congestion_detector.go @@ -121,8 +121,8 @@ var ( MinBytesRatio: 0.5, MinDurationRatio: 0.5, - JQRMinDelay: 15 * time.Millisecond, - DQRMaxDelay: 5 * time.Millisecond, + JQRMinDelay: 40 * time.Millisecond, + DQRMaxDelay: 15 * time.Millisecond, WeightedLoss: defaultWeightedLossConfig, JQRMinWeightedLoss: 0.25, @@ -228,34 +228,30 @@ func (q *qdMeasurement) ProcessPacketGroup(pg *packetGroup, groupIdx int) { } q.maxSendTime = max(q.maxSendTime, maxSendTime) - if pqd < q.dqrMax { + switch { + case pqd < q.dqrMax: q.numDQRGroups++ if q.numJQRGroups > 0 { - // JQR continuity is broken + // broken continuity, seal q.isSealed = true - return - } - - if q.dqrConfig.IsTriggered(q.numDQRGroups, q.maxSendTime-q.minSendTime) { + } else if q.dqrConfig.IsTriggered(q.numDQRGroups, q.maxSendTime-q.minSendTime) { q.isSealed = true q.queuingRegion = queuingRegionDQR - return } - } - if pqd > q.jqrMin { + case pqd > q.jqrMin: q.numJQRGroups++ if q.numDQRGroups > 0 { - // DQR continuity is broken + // broken continuity, seal q.isSealed = true - return - } - - if q.jqrConfig.IsTriggered(q.numJQRGroups, q.maxSendTime-q.minSendTime) { + } else if q.jqrConfig.IsTriggered(q.numJQRGroups, q.maxSendTime-q.minSendTime) { q.isSealed = true q.queuingRegion = queuingRegionJQR - return } + + default: + // broken continuity, seal + q.isSealed = true } } @@ -351,17 +347,19 @@ func (l *lossMeasurement) ProcessPacketGroup(pg *packetGroup, groupIdx int) { if weightedLoss > l.jqrMinLoss { l.weightedLoss = weightedLoss l.queuingRegion = queuingRegionJQR - l.isDQRSealed = true // seal DQR also as JQR is already hit + l.isDQRSealed = true // seal DQR also as queuing region has been determined return } } - if l.dqrConfig.IsTriggered(l.numGroups, l.ts.Duration()) { + if !l.isDQRSealed && l.dqrConfig.IsTriggered(l.numGroups, l.ts.Duration()) { l.isDQRSealed = true - l.weightedLoss = l.ts.WeightedLoss() - if l.weightedLoss < l.dqrMaxLoss { + weightedLoss := l.ts.WeightedLoss() + if weightedLoss < l.dqrMaxLoss { + l.weightedLoss = weightedLoss l.queuingRegion = queuingRegionDQR + l.isJQRSealed = true // seal JQR also as queuing region has been determined return } } @@ -452,8 +450,8 @@ var ( ProbeRegulator: ccutils.DefaultProbeRegulatorConfig, ProbeSignal: defaultProbeSignalConfig, - JQRMinDelay: 15 * time.Millisecond, - DQRMaxDelay: 5 * time.Millisecond, + JQRMinDelay: 40 * time.Millisecond, + DQRMaxDelay: 15 * time.Millisecond, WeightedLoss: defaultWeightedLossConfig, JQRMinWeightedLoss: 0.25, @@ -652,6 +650,23 @@ func (c *congestionDetector) HandleTWCCFeedback(report *rtcp.TransportLayerCC) { sequenceNumber := report.BaseSequenceNumber endSequenceNumberExclusive := sequenceNumber + report.PacketStatusCount deltaIdx := 0 + processSymbol := func(symbol uint16) { + recvTime := int64(0) + isLost := false + if symbol != rtcp.TypeTCCPacketNotReceived { + recvRefTime += report.RecvDeltas[deltaIdx].Delta + deltaIdx++ + + recvTime = recvRefTime + } else { + isLost = true + } + pi, sendDelta, recvDelta := c.packetTracker.RecordPacketIndicationFromRemote(sequenceNumber, recvTime) + if pi.sendTime != 0 { + trackPacketGroup(&pi, sendDelta, recvDelta, isLost) + } + sequenceNumber++ + } for _, chunk := range report.PacketChunks { if sequenceNumber == endSequenceNumberExclusive { break @@ -664,21 +679,7 @@ func (c *congestionDetector) HandleTWCCFeedback(report *rtcp.TransportLayerCC) { break } - recvTime := int64(0) - isLost := false - if chunk.PacketStatusSymbol != rtcp.TypeTCCPacketNotReceived { - recvRefTime += report.RecvDeltas[deltaIdx].Delta - deltaIdx++ - - recvTime = recvRefTime - } else { - isLost = true - } - pi, sendDelta, recvDelta := c.packetTracker.RecordPacketIndicationFromRemote(sequenceNumber, recvTime) - if pi.sendTime != 0 { - trackPacketGroup(&pi, sendDelta, recvDelta, isLost) - } - sequenceNumber++ + processSymbol(chunk.PacketStatusSymbol) } case *rtcp.StatusVectorChunk: @@ -687,21 +688,7 @@ func (c *congestionDetector) HandleTWCCFeedback(report *rtcp.TransportLayerCC) { break } - recvTime := int64(0) - isLost := false - if symbol != rtcp.TypeTCCPacketNotReceived { - recvRefTime += report.RecvDeltas[deltaIdx].Delta - deltaIdx++ - - recvTime = recvRefTime - } else { - isLost = true - } - pi, sendDelta, recvDelta := c.packetTracker.RecordPacketIndicationFromRemote(sequenceNumber, recvTime) - if pi.sendTime != 0 { - trackPacketGroup(&pi, sendDelta, recvDelta, isLost) - } - sequenceNumber++ + processSymbol(symbol) } } } @@ -855,7 +842,6 @@ func (c *congestionDetector) prunePacketGroups() { } func (c *congestionDetector) updateCongestionSignal( - stage string, qdJQRConfig CongestionSignalConfig, qdDQRConfig CongestionSignalConfig, lossJQRConfig CongestionSignalConfig, @@ -902,7 +888,6 @@ func (c *congestionDetector) updateCongestionSignal( func (c *congestionDetector) updateEarlyWarningSignal() { c.updateCongestionSignal( - "early-warning", c.params.Config.QueuingDelayEarlyWarningJQR, c.params.Config.QueuingDelayEarlyWarningDQR, c.params.Config.LossEarlyWarningJQR, @@ -912,7 +897,6 @@ func (c *congestionDetector) updateEarlyWarningSignal() { func (c *congestionDetector) updateCongestedSignal() { c.updateCongestionSignal( - "congested", c.params.Config.QueuingDelayCongestedJQR, c.params.Config.QueuingDelayCongestedDQR, c.params.Config.LossCongestedJQR, diff --git a/pkg/sfu/bwe/sendsidebwe/packet_group.go b/pkg/sfu/bwe/sendsidebwe/packet_group.go index 0c9f2d0f5cf..b64727fdfd3 100644 --- a/pkg/sfu/bwe/sendsidebwe/packet_group.go +++ b/pkg/sfu/bwe/sendsidebwe/packet_group.go @@ -182,7 +182,7 @@ func (p *packetGroup) Add(pi *packetInfo, sendDelta, recvDelta int64, isLost boo p.maxRecvTime = max(p.maxRecvTime, pi.recvTime) p.acked.add(int(pi.size), pi.isRTX, pi.isProbe) - if p.snBitmap.IsSet(pi.sequenceNumber - p.minSequenceNumber) { + if int(pi.sequenceNumber-p.minSequenceNumber) < p.snBitmap.Len() && p.snBitmap.IsSet(pi.sequenceNumber-p.minSequenceNumber) { // an earlier packet reported as lost has been received p.snBitmap.Clear(pi.sequenceNumber - p.minSequenceNumber) p.lost.remove(int(pi.size), pi.isRTX, pi.isProbe) diff --git a/pkg/sfu/codecmunger/vp8.go b/pkg/sfu/codecmunger/vp8.go index 3c89d2e1ac4..696c355b1a3 100644 --- a/pkg/sfu/codecmunger/vp8.go +++ b/pkg/sfu/codecmunger/vp8.go @@ -226,7 +226,7 @@ func (v *VP8) UpdateAndGet(extPkt *buffer.ExtPacket, snOutOfOrder bool, snHasGap // which layer the missing packets belong to. A layer could have multiple packets. So, keep track // of pictures that are forwarded even though they will be filtered out based on temporal layer // requirements. That allows forwarding of the complete picture. - if vp8.T && vp8.TID > uint8(maxTemporalLayer) { + if extPkt.Temporal > maxTemporalLayer { v.exemptedPictureIds.Set(extPictureId, true) // trim cache if necessary for v.exemptedPictureIds.Len() > exemptedPictureIdsThreshold { @@ -235,7 +235,7 @@ func (v *VP8) UpdateAndGet(extPkt *buffer.ExtPacket, snOutOfOrder bool, snHasGap } } } else { - if vp8.T && vp8.TID > uint8(maxTemporalLayer) { + if extPkt.Temporal > maxTemporalLayer { // drop only if not exempted _, ok := v.exemptedPictureIds.Get(extPictureId) if !ok { diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 1c5c3c682fe..abd06f04f38 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -874,7 +874,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { } // add extensions - if tp.ddBytes != nil { + if d.dependencyDescriptorExtID != 0 && tp.ddBytes != nil { hdr.SetExtension(uint8(d.dependencyDescriptorExtID), tp.ddBytes) } if d.playoutDelayExtID != 0 && d.playoutDelay != nil { @@ -1008,7 +1008,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa // Hold sending padding packets till first RTCP-RR is received for this RTP stream. // That is definitive proof that the remote side knows about this RTP stream. - if d.rtpStats.LastReceiverReportTime().IsZero() && !paddingOnMute { + if d.rtpStats.LastReceiverReportTime() == 0 && !paddingOnMute { return 0 } @@ -2093,7 +2093,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int { if !d.writable.Load() || !d.rtpStats.IsActive() || (d.absSendTimeExtID == 0 && d.transportWideExtID == 0) || - d.rtpStats.LastReceiverReportTime().IsZero() || + d.rtpStats.LastReceiverReportTime() == 0 || d.sequencer == nil { return 0 } @@ -2261,7 +2261,7 @@ func (d *DownTrack) GetDeltaStatsSender() map[uint32]*buffer.StreamStatsWithLaye } func (d *DownTrack) GetPrimaryStreamLastReceiverReportTime() time.Time { - return d.rtpStats.LastReceiverReportTime() + return time.Unix(0, d.rtpStats.LastReceiverReportTime()) } func (d *DownTrack) GetPrimaryStreamPacketsSent() uint64 { diff --git a/pkg/sfu/rtpstats/rtpstats_base.go b/pkg/sfu/rtpstats/rtpstats_base.go index 40ec921c30c..d42a40df80a 100644 --- a/pkg/sfu/rtpstats/rtpstats_base.go +++ b/pkg/sfu/rtpstats/rtpstats_base.go @@ -24,6 +24,7 @@ import ( "github.com/livekit/mediatransportutil" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/utils" + "github.com/livekit/protocol/utils/mono" ) const ( @@ -298,7 +299,7 @@ func (r *rtpStatsBase) newSnapshotID(extStartSN uint64) uint32 { } if r.initialized { - r.snapshots[id-cFirstSnapshotID] = initSnapshot(time.Now(), extStartSN) + r.snapshots[id-cFirstSnapshotID] = initSnapshot(mono.UnixNano(), extStartSN) } return id } @@ -307,7 +308,7 @@ func (r *rtpStatsBase) UpdateFir(firCount uint32) { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -318,7 +319,7 @@ func (r *rtpStatsBase) UpdateFirTime() { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -329,7 +330,7 @@ func (r *rtpStatsBase) UpdateKeyFrame(kfCount uint32) { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -341,7 +342,7 @@ func (r *rtpStatsBase) UpdateRtt(rtt uint32) { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -366,7 +367,8 @@ func (r *rtpStatsBase) GetRtt() uint32 { } func (r *rtpStatsBase) maybeAdjustFirstPacketTime(srData *livekit.RTCPSenderReportState, tsOffset uint64, extStartTS uint64) (err error, loggingFields []interface{}) { - if time.Since(r.startTime) > cFirstPacketTimeAdjustWindow { + nowNano := mono.UnixNano() + if time.Duration(nowNano-r.startTime) > cFirstPacketTimeAdjustWindow { return } @@ -376,7 +378,7 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime(srData *livekit.RTCPSenderRepo // abnormal delay (maybe due to pacing or maybe due to queuing // in some network element along the way), push back first time // to an earlier instance. - timeSinceReceive := time.Since(time.Unix(0, srData.AtAdjusted)) + timeSinceReceive := time.Duration(nowNano - srData.AtAdjusted) extNowTS := srData.RtpTimestampExt - tsOffset + uint64(timeSinceReceive.Nanoseconds()*int64(r.params.ClockRate)/1e9) samplesDiff := int64(extNowTS - extStartTS) if samplesDiff < 0 { @@ -385,13 +387,13 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime(srData *livekit.RTCPSenderRepo } samplesDuration := time.Duration(float64(samplesDiff) / float64(r.params.ClockRate) * float64(time.Second)) - timeSinceFirst := time.Since(time.Unix(0, r.firstTime)) + timeSinceFirst := time.Duration(nowNano - r.firstTime) now := r.firstTime + timeSinceFirst.Nanoseconds() firstTime := now - samplesDuration.Nanoseconds() getFields := func() []interface{} { return []interface{}{ - "startTime", r.startTime, + "startTime", time.Unix(0, r.startTime), "nowTime", time.Unix(0, now), "before", time.Unix(0, r.firstTime), "after", time.Unix(0, firstTime), @@ -455,7 +457,7 @@ func (r *rtpStatsBase) deltaInfo( "snapshotID", snapshotID, "snapshotNow", now, "snapshotThen", then, - "duration", endTime.Sub(startTime), + "duration", time.Duration(endTime - startTime), "packetsExpected", packetsExpected, } err = errors.New("too many packets expected in delta") @@ -463,8 +465,8 @@ func (r *rtpStatsBase) deltaInfo( } if packetsExpected == 0 { deltaInfo = &RTPDeltaInfo{ - StartTime: startTime, - EndTime: endTime, + StartTime: time.Unix(0, startTime), + EndTime: time.Unix(0, endTime), } return } @@ -481,7 +483,7 @@ func (r *rtpStatsBase) deltaInfo( "snapshotID", snapshotID, "snapshotNow", now, "snapshotThen", then, - "duration", endTime.Sub(startTime), + "duration", time.Duration(endTime - startTime), "packetsExpected", packetsExpected, "packetsPadding", packetsPadding, "packetsLost", packetsLost, @@ -493,8 +495,8 @@ func (r *rtpStatsBase) deltaInfo( } deltaInfo = &RTPDeltaInfo{ - StartTime: startTime, - EndTime: endTime, + StartTime: time.Unix(0, startTime), + EndTime: time.Unix(0, endTime), Packets: uint32(packetsExpected), Bytes: now.bytes - then.bytes, HeaderBytes: now.headerBytes - then.headerBytes, @@ -667,7 +669,7 @@ func (r *rtpStatsBase) getAndResetSnapshot(snapshotID uint32, extStartSN uint64, } // snapshot now - now := r.getSnapshot(time.Now(), extHighestSN+1) + now := r.getSnapshot(mono.UnixNano(), extHighestSN+1) r.snapshots[idx] = now return &then, &now } @@ -765,7 +767,7 @@ func (r *rtpStatsBase) updateGapHistogram(gap int) { } } -func (r *rtpStatsBase) getSnapshot(startTime time.Time, extStartSN uint64) snapshot { +func (r *rtpStatsBase) getSnapshot(startTime int64, extStartSN uint64) snapshot { return snapshot{ snapshotLite: r.getSnapshotLite(startTime, extStartSN), headerBytes: r.headerBytes, @@ -785,7 +787,7 @@ func (r *rtpStatsBase) getSnapshot(startTime time.Time, extStartSN uint64) snaps // ---------------------------------- -func initSnapshot(startTime time.Time, extStartSN uint64) snapshot { +func initSnapshot(startTime int64, extStartSN uint64) snapshot { return snapshot{ snapshotLite: initSnapshotLite(startTime, extStartSN), } @@ -800,8 +802,8 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo { return nil } - startTime := time.Time{} - endTime := time.Time{} + startTime := int64(0) + endTime := int64(0) packets := uint32(0) bytes := uint64(0) @@ -833,12 +835,12 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo { continue } - if startTime.IsZero() || startTime.After(deltaInfo.StartTime) { - startTime = deltaInfo.StartTime + if startTime == 0 || startTime > deltaInfo.StartTime.UnixNano() { + startTime = deltaInfo.StartTime.UnixNano() } - if endTime.IsZero() || endTime.Before(deltaInfo.EndTime) { - endTime = deltaInfo.EndTime + if endTime == 0 || endTime < deltaInfo.EndTime.UnixNano() { + endTime = deltaInfo.EndTime.UnixNano() } packets += deltaInfo.Packets @@ -871,13 +873,13 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo { plis += deltaInfo.Plis firs += deltaInfo.Firs } - if startTime.IsZero() || endTime.IsZero() { + if startTime == 0 || endTime == 0 { return nil } return &RTPDeltaInfo{ - StartTime: startTime, - EndTime: endTime, + StartTime: time.Unix(0, startTime), + EndTime: time.Unix(0, endTime), Packets: packets, Bytes: bytes, HeaderBytes: headerBytes, diff --git a/pkg/sfu/rtpstats/rtpstats_base_lite.go b/pkg/sfu/rtpstats/rtpstats_base_lite.go index cff08e07891..6e30a6702ac 100644 --- a/pkg/sfu/rtpstats/rtpstats_base_lite.go +++ b/pkg/sfu/rtpstats/rtpstats_base_lite.go @@ -22,6 +22,7 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils/mono" "go.uber.org/zap/zapcore" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -64,7 +65,7 @@ func (r *RTPDeltaInfoLite) MarshalLogObject(e zapcore.ObjectEncoder) error { type snapshotLite struct { isValid bool - startTime time.Time + startTime int64 extStartSN uint64 bytes uint64 @@ -82,7 +83,7 @@ func (s *snapshotLite) MarshalLogObject(e zapcore.ObjectEncoder) error { } e.AddBool("isValid", s.isValid) - e.AddTime("startTime", s.startTime) + e.AddTime("startTime", time.Unix(0, s.startTime)) e.AddUint64("extStartSN", s.extStartSN) e.AddUint64("bytes", s.bytes) e.AddUint64("packetsOutOfOrder", s.packetsOutOfOrder) @@ -106,8 +107,8 @@ type rtpStatsBaseLite struct { initialized bool - startTime time.Time - endTime time.Time + startTime int64 + endTime int64 bytes uint64 @@ -123,7 +124,7 @@ type rtpStatsBaseLite struct { nackRepeated uint32 plis uint32 - lastPli time.Time + lastPli int64 nextSnapshotLiteID uint32 snapshotLites []snapshotLite @@ -178,7 +179,7 @@ func (r *rtpStatsBaseLite) Stop() { r.lock.Lock() defer r.lock.Unlock() - r.endTime = time.Now() + r.endTime = mono.UnixNano() } func (r *rtpStatsBaseLite) newSnapshotLiteID(extStartSN uint64) uint32 { @@ -192,7 +193,7 @@ func (r *rtpStatsBaseLite) newSnapshotLiteID(extStartSN uint64) uint32 { } if r.initialized { - r.snapshotLites[id-cFirstSnapshotID] = initSnapshotLite(time.Now(), extStartSN) + r.snapshotLites[id-cFirstSnapshotID] = initSnapshotLite(mono.UnixNano(), extStartSN) } return id } @@ -201,14 +202,14 @@ func (r *rtpStatsBaseLite) IsActive() bool { r.lock.RLock() defer r.lock.RUnlock() - return r.initialized && r.endTime.IsZero() + return r.initialized && r.endTime == 0 } func (r *rtpStatsBaseLite) UpdateNack(nackCount uint32) { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -219,7 +220,7 @@ func (r *rtpStatsBaseLite) UpdateNackProcessed(nackAckCount uint32, nackMissCoun r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -232,7 +233,7 @@ func (r *rtpStatsBaseLite) CheckAndUpdatePli(throttle int64, force bool) bool { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() || (!force && time.Now().UnixNano()-r.lastPli.UnixNano() < throttle) { + if r.endTime != 0 || (!force && mono.UnixNano()-r.lastPli < throttle) { return false } r.updatePliLocked(1) @@ -244,7 +245,7 @@ func (r *rtpStatsBaseLite) UpdatePliAndTime(pliCount uint32) { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -256,7 +257,7 @@ func (r *rtpStatsBaseLite) UpdatePli(pliCount uint32) { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -271,7 +272,7 @@ func (r *rtpStatsBaseLite) UpdatePliTime() { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -279,10 +280,10 @@ func (r *rtpStatsBaseLite) UpdatePliTime() { } func (r *rtpStatsBaseLite) updatePliTimeLocked() { - r.lastPli = time.Now() + r.lastPli = mono.UnixNano() } -func (r *rtpStatsBaseLite) LastPli() time.Time { +func (r *rtpStatsBaseLite) LastPli() int64 { r.lock.RLock() defer r.lock.RUnlock() @@ -322,15 +323,15 @@ func (r *rtpStatsBaseLite) deltaInfoLite( "snapshotLiteNow", now, "snapshotLiteThen", then, "packetsExpected", packetsExpected, - "duration", endTime.Sub(startTime).String(), + "duration", time.Duration(endTime - startTime), } err = errors.New("too many packets expected in delta lite") return } if packetsExpected == 0 { deltaInfoLite = &RTPDeltaInfoLite{ - StartTime: startTime, - EndTime: endTime, + StartTime: time.Unix(0, startTime), + EndTime: time.Unix(0, endTime), } return } @@ -346,14 +347,14 @@ func (r *rtpStatsBaseLite) deltaInfoLite( "snapshotLiteThen", then, "packetsExpected", packetsExpected, "packetsLost", packetsLost, - "duration", endTime.Sub(startTime).String(), + "duration", time.Duration(endTime - startTime), } err = errors.New("unexpected number of packets lost in delta lite") } deltaInfoLite = &RTPDeltaInfoLite{ - StartTime: startTime, - EndTime: endTime, + StartTime: time.Unix(0, startTime), + EndTime: time.Unix(0, endTime), Packets: packetsExpected, Bytes: now.bytes - then.bytes, PacketsLost: packetsLost, @@ -369,17 +370,17 @@ func (r *rtpStatsBaseLite) marshalLogObject(e zapcore.ObjectEncoder, packetsExpe } endTime := r.endTime - if endTime.IsZero() { - endTime = time.Now() + if endTime == 0 { + endTime = mono.UnixNano() } - elapsed := endTime.Sub(r.startTime) + elapsed := time.Duration(endTime - r.startTime) if elapsed == 0 { return 0, errors.New("no time elapsed") } elapsedSeconds := elapsed.Seconds() - e.AddTime("startTime", r.startTime) - e.AddTime("endTime", r.endTime) + e.AddTime("startTime", time.Unix(0, r.startTime)) + e.AddTime("endTime", time.Unix(0, r.endTime)) e.AddDuration("elapsed", elapsed) e.AddUint64("packetsExpected", packetsExpected) @@ -424,20 +425,20 @@ func (r *rtpStatsBaseLite) marshalLogObject(e zapcore.ObjectEncoder, packetsExpe e.AddUint32("nackRepeated", r.nackRepeated) e.AddUint32("plis", r.plis) - e.AddTime("lastPli", r.lastPli) + e.AddTime("lastPli", time.Unix(0, r.lastPli)) return elapsedSeconds, nil } func (r *rtpStatsBaseLite) toProto(packetsExpected, packetsSeenMinusPadding, packetsLost uint64) *livekit.RTPStats { - if r.startTime.IsZero() { + if r.startTime == 0 { return nil } endTime := r.endTime - if endTime.IsZero() { - endTime = time.Now() + if endTime == 0 { + endTime = mono.UnixNano() } - elapsed := endTime.Sub(r.startTime).Seconds() + elapsed := time.Duration(endTime - r.startTime).Seconds() if elapsed == 0.0 { return nil } @@ -452,8 +453,8 @@ func (r *rtpStatsBaseLite) toProto(packetsExpected, packetsSeenMinusPadding, pac } p := &livekit.RTPStats{ - StartTime: timestamppb.New(r.startTime), - EndTime: timestamppb.New(endTime), + StartTime: timestamppb.New(time.Unix(0, r.startTime)), + EndTime: timestamppb.New(time.Unix(0, endTime)), Duration: elapsed, Packets: uint32(packetsSeenMinusPadding), PacketRate: packetRate, @@ -468,7 +469,7 @@ func (r *rtpStatsBaseLite) toProto(packetsExpected, packetsSeenMinusPadding, pac NackMisses: r.nackMisses, NackRepeated: r.nackRepeated, Plis: r.plis, - LastPli: timestamppb.New(r.lastPli), + LastPli: timestamppb.New(time.Unix(0, r.lastPli)), } gapsPresent := false @@ -508,7 +509,7 @@ func (r *rtpStatsBaseLite) getAndResetSnapshotLite(snapshotLiteID uint32, extSta } // snapshot now - now := r.getSnapshotLite(time.Now(), extHighestSN+1) + now := r.getSnapshotLite(mono.UnixNano(), extHighestSN+1) r.snapshotLites[idx] = now return &then, &now } @@ -526,7 +527,7 @@ func (r *rtpStatsBaseLite) updateGapHistogram(gap int) { } } -func (r *rtpStatsBaseLite) getSnapshotLite(startTime time.Time, extStartSN uint64) snapshotLite { +func (r *rtpStatsBaseLite) getSnapshotLite(startTime int64, extStartSN uint64) snapshotLite { return snapshotLite{ isValid: true, startTime: startTime, @@ -540,7 +541,7 @@ func (r *rtpStatsBaseLite) getSnapshotLite(startTime time.Time, extStartSN uint6 // ---------------------------------- -func initSnapshotLite(startTime time.Time, extStartSN uint64) snapshotLite { +func initSnapshotLite(startTime int64, extStartSN uint64) snapshotLite { return snapshotLite{ isValid: true, startTime: startTime, diff --git a/pkg/sfu/rtpstats/rtpstats_receiver.go b/pkg/sfu/rtpstats/rtpstats_receiver.go index a3dd7936e2c..95d92d49376 100644 --- a/pkg/sfu/rtpstats/rtpstats_receiver.go +++ b/pkg/sfu/rtpstats/rtpstats_receiver.go @@ -27,6 +27,7 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" protoutils "github.com/livekit/protocol/utils" + "github.com/livekit/protocol/utils/mono" ) const ( @@ -137,7 +138,7 @@ func (r *RTPStatsReceiver) Update( r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { flowState.IsNotHandled = true return } @@ -181,7 +182,7 @@ func (r *RTPStatsReceiver) Update( r.initialized = true - r.startTime = time.Now() + r.startTime = mono.UnixNano() r.firstTime = packetTime r.highestTime = packetTime @@ -275,14 +276,13 @@ func (r *RTPStatsReceiver) Update( } if r.isInRange(resSN.ExtendedVal, resSN.PreExtendedHighest) { - if r.history.IsSet(resSN.ExtendedVal) { + if r.history.GetAndSet(resSN.ExtendedVal) { r.bytesDuplicate += pktSize r.headerBytesDuplicate += uint64(hdrSize) r.packetsDuplicate++ flowState.IsDuplicate = true } else { r.packetsLost-- - r.history.Set(resSN.ExtendedVal) } } @@ -461,14 +461,15 @@ func (r *RTPStatsReceiver) checkRTPClockSkewAgainstMediaPathForSenderReport(srDa return } - timeSinceSR := time.Since(time.Unix(0, srData.AtAdjusted)) + nowNano := mono.UnixNano() + timeSinceSR := time.Duration(nowNano - srData.AtAdjusted) extNowTSSR := srData.RtpTimestampExt + uint64(timeSinceSR.Nanoseconds()*int64(r.params.ClockRate)/1e9) - timeSinceHighest := time.Since(time.Unix(0, r.highestTime)) + timeSinceHighest := time.Duration(nowNano - r.highestTime) extNowTSHighest := r.timestamp.GetExtendedHighest() + uint64(timeSinceHighest.Nanoseconds()*int64(r.params.ClockRate)/1e9) diffHighest := extNowTSSR - extNowTSHighest - timeSinceFirst := time.Since(time.Unix(0, r.firstTime)) + timeSinceFirst := time.Duration(nowNano - r.firstTime) extNowTSFirst := r.timestamp.GetExtendedStart() + uint64(timeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) diffFirst := extNowTSSR - extNowTSFirst diff --git a/pkg/sfu/rtpstats/rtpstats_receiver_lite.go b/pkg/sfu/rtpstats/rtpstats_receiver_lite.go index 5f6ef29c7cd..84812a323ff 100644 --- a/pkg/sfu/rtpstats/rtpstats_receiver_lite.go +++ b/pkg/sfu/rtpstats/rtpstats_receiver_lite.go @@ -15,12 +15,11 @@ package rtpstats import ( - "time" - "go.uber.org/zap/zapcore" "github.com/livekit/livekit-server/pkg/sfu/utils" "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/utils/mono" ) type RTPFlowStateLite struct { @@ -70,7 +69,7 @@ func (r *RTPStatsReceiverLite) Update(packetTime int64, packetSize int, sequence r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { flowStateLite.IsNotHandled = true return } @@ -79,7 +78,7 @@ func (r *RTPStatsReceiverLite) Update(packetTime int64, packetSize int, sequence if !r.initialized { r.initialized = true - r.startTime = time.Now() + r.startTime = mono.UnixNano() resSN = r.sequenceNumber.Update(sequenceNumber) diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index e3ee972a884..c38116eaafe 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -26,6 +26,7 @@ import ( "github.com/livekit/mediatransportutil" "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils/mono" ) @@ -121,7 +122,7 @@ func (w wrappedReceptionReportsLogger) MarshalLogObject(e zapcore.ObjectEncoder) type senderSnapshot struct { isValid bool - startTime time.Time + startTime int64 extStartSN uint64 bytes uint64 @@ -151,10 +152,11 @@ type senderSnapshot struct { maxJitterFeed float64 maxJitter float64 - extLastRRSN uint64 - intervalStats intervalStats - processedReceptionReports []rtcp.ReceptionReport - skippedReceptionReports []rtcp.ReceptionReport + extLastRRSN uint64 + intervalStats intervalStats + processedReceptionReports []rtcp.ReceptionReport + skippedReceptionReports []rtcp.ReceptionReport + metadataCacheOverflowCount int } func (s *senderSnapshot) MarshalLogObject(e zapcore.ObjectEncoder) error { @@ -163,7 +165,7 @@ func (s *senderSnapshot) MarshalLogObject(e zapcore.ObjectEncoder) error { } e.AddBool("isValid", s.isValid) - e.AddTime("startTime", s.startTime) + e.AddTime("startTime", time.Unix(0, s.startTime)) e.AddUint64("extStartSN", s.extStartSN) e.AddUint64("bytes", s.bytes) e.AddUint64("headerBytes", s.headerBytes) @@ -187,6 +189,7 @@ func (s *senderSnapshot) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddObject("intervalStats", &s.intervalStats) e.AddObject("processedReceptionReports", wrappedReceptionReportsLogger{s, false}) e.AddObject("skippedReceptionReports", wrappedReceptionReportsLogger{s, true}) + e.AddInt("metadataCacheOverflowCount", s.metadataCacheOverflowCount) return nil } @@ -208,7 +211,7 @@ type RTPStatsSender struct { rttMarker rttMarker - lastRRTime time.Time + lastRRTime int64 lastRR rtcp.ReceptionReport extStartTS uint64 @@ -227,11 +230,10 @@ type RTPStatsSender struct { nextSenderSnapshotID uint32 senderSnapshots []senderSnapshot - clockSkewCount int - metadataCacheOverflowCount int - largeJumpNegativeCount int - largeJumpCount int - timeReversedCount int + clockSkewCount int + largeJumpNegativeCount int + largeJumpCount int + timeReversedCount int } func NewRTPStatsSender(params RTPStatsParams, cacheSize int) *RTPStatsSender { @@ -300,7 +302,7 @@ func (r *RTPStatsSender) NewSenderSnapshotId() uint32 { } if r.initialized { - r.senderSnapshots[id-cFirstSnapshotID] = initSenderSnapshot(time.Now(), r.extHighestSN) + r.senderSnapshots[id-cFirstSnapshotID] = initSenderSnapshot(mono.UnixNano(), r.extHighestSN) } return id } @@ -318,7 +320,7 @@ func (r *RTPStatsSender) Update( r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -330,7 +332,7 @@ func (r *RTPStatsSender) Update( r.initialized = true - r.startTime = time.Now() + r.startTime = mono.UnixNano() r.highestTime = packetTime @@ -363,18 +365,21 @@ func (r *RTPStatsSender) Update( pktSize := uint64(hdrSize + payloadSize + paddingSize) isDuplicate := false gapSN := int64(extSequenceNumber - r.extHighestSN) - logger := r.logger.WithUnlikelyValues( - "currSN", extSequenceNumber, - "gapSN", gapSN, - "currTS", extTimestamp, - "gapTS", int64(extTimestamp-r.extHighestTS), - "packetTime", packetTime, - "marker", marker, - "hdrSize", hdrSize, - "payloadSize", payloadSize, - "paddingSize", paddingSize, - "rtpStats", lockedRTPStatsSenderLogEncoder{r}, - ) + ulgr := func() logger.UnlikelyLogger { + return r.logger.WithUnlikelyValues( + "currSN", extSequenceNumber, + "gapSN", gapSN, + "currTS", extTimestamp, + "gapTS", int64(extTimestamp-r.extHighestTS), + "packetTime", time.Unix(0, packetTime), + "timeSinceHighest", time.Duration(packetTime-r.highestTime), + "marker", marker, + "hdrSize", hdrSize, + "payloadSize", payloadSize, + "paddingSize", paddingSize, + "rtpStats", lockedRTPStatsSenderLogEncoder{r}, + ) + } if gapSN <= 0 { // duplicate OR out-of-order if payloadSize == 0 && extSequenceNumber < r.extStartSN { // do not start on a padding only packet @@ -382,7 +387,7 @@ func (r *RTPStatsSender) Update( } if extSequenceNumber < r.extStartSN { - r.packetsLost += r.extStartSN - extSequenceNumber + r.packetsLost += r.extStartSN - extSequenceNumber - 1 // adjust start of snapshots for i := uint32(0); i < r.nextSnapshotID-cFirstSnapshotID; i++ { @@ -401,7 +406,7 @@ func (r *RTPStatsSender) Update( } } - logger.Infow( + ulgr().Infow( "adjusting start sequence number", "snAfter", extSequenceNumber, "tsAfter", extTimestamp, @@ -426,7 +431,7 @@ func (r *RTPStatsSender) Update( if !isDuplicate && -gapSN >= cSequenceNumberLargeJumpThreshold { r.largeJumpNegativeCount++ if (r.largeJumpNegativeCount-1)%100 == 0 { - logger.Warnw( + ulgr().Warnw( "large sequence number gap negative", nil, "count", r.largeJumpNegativeCount, ) @@ -436,7 +441,7 @@ func (r *RTPStatsSender) Update( if gapSN >= cSequenceNumberLargeJumpThreshold { r.largeJumpCount++ if (r.largeJumpCount-1)%100 == 0 { - logger.Warnw( + ulgr().Warnw( "large sequence number gap", nil, "count", r.largeJumpCount, ) @@ -446,7 +451,7 @@ func (r *RTPStatsSender) Update( if extTimestamp < r.extHighestTS { r.timeReversedCount++ if (r.timeReversedCount-1)%100 == 0 { - logger.Warnw( + ulgr().Warnw( "time reversed", nil, "count", r.timeReversedCount, ) @@ -466,7 +471,7 @@ func (r *RTPStatsSender) Update( } if extTimestamp < r.extStartTS { - logger.Infow( + ulgr().Infow( "adjusting start timestamp", "snAfter", extSequenceNumber, "tsAfter", extTimestamp, @@ -512,7 +517,7 @@ func (r *RTPStatsSender) UpdateLayerLockPliAndTime(pliCount uint32) { r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } @@ -531,12 +536,12 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt r.lock.Lock() defer r.lock.Unlock() - if !r.initialized || !r.endTime.IsZero() { + if !r.initialized || r.endTime != 0 { return } extHighestSNFromRR := r.extHighestSNFromRR&0xFFFF_FFFF_0000_0000 + uint64(rr.LastSequenceNumber) - if !r.lastRRTime.IsZero() { + if r.lastRRTime != 0 { if (rr.LastSequenceNumber-r.lastRR.LastSequenceNumber) < (1<<31) && rr.LastSequenceNumber < r.lastRR.LastSequenceNumber { extHighestSNFromRR += (1 << 32) } @@ -547,10 +552,10 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt return } - if !r.lastRRTime.IsZero() && r.extHighestSNFromRR > extHighestSNFromRR { + if r.lastRRTime != 0 && r.extHighestSNFromRR > extHighestSNFromRR { r.logger.Debugw( fmt.Sprintf("receiver report potentially out of order, highestSN: existing: %d, received: %d", r.extHighestSNFromRR, extHighestSNFromRR), - "sinceLastRR", time.Since(r.lastRRTime), + "sinceLastRR", time.Duration(mono.UnixNano()-r.lastRRTime), "receivedRR", rr, "rtpStats", lockedRTPStatsSenderLogEncoder{r}, ) @@ -606,9 +611,9 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt } if int64(extReceivedRRSN-s.extLastRRSN) < 0 || (extReceivedRRSN-s.extLastRRSN) > (1<<15) { - timeSinceLastRR := time.Since(r.lastRRTime) - if r.lastRRTime.IsZero() { - timeSinceLastRR = time.Since(r.startTime) + timeSinceLastRR := time.Duration(mono.UnixNano() - r.lastRRTime) + if r.lastRRTime == 0 { + timeSinceLastRR = time.Duration(mono.UnixNano() - r.startTime) } r.logger.Infow( "rr interval too big, skipping", @@ -628,21 +633,22 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt eis := &s.intervalStats eis.aggregate(&is) if is.packetsNotFoundMetadata != 0 { - timeSinceLastRR := time.Since(r.lastRRTime) - if r.lastRRTime.IsZero() { - timeSinceLastRR = time.Since(r.startTime) + timeSinceLastRR := time.Duration(mono.UnixNano() - r.lastRRTime) + if r.lastRRTime == 0 { + timeSinceLastRR = time.Duration(mono.UnixNano() - r.startTime) } - r.metadataCacheOverflowCount++ - if (r.metadataCacheOverflowCount-1)%10 == 0 { + s.metadataCacheOverflowCount++ + if (s.metadataCacheOverflowCount-1)%10 == 0 { r.logger.Infow( "metadata cache overflow", + "senderSnapshotID", i+cFirstSnapshotID, "timeSinceLastRR", timeSinceLastRR, "receivedRR", rr, "extReceivedRRSN", extReceivedRRSN, "packetsInInterval", extReceivedRRSN-s.extLastRRSN, "intervalStats", &is, "aggregateIntervalStats", eis, - "count", r.metadataCacheOverflowCount, + "count", s.metadataCacheOverflowCount, "rtpStats", lockedRTPStatsSenderLogEncoder{r}, ) } @@ -651,12 +657,12 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt s.processedReceptionReports = append(s.processedReceptionReports, rr) } - r.lastRRTime = time.Now() + r.lastRRTime = mono.UnixNano() r.lastRR = rr return } -func (r *RTPStatsSender) LastReceiverReportTime() time.Time { +func (r *RTPStatsSender) LastReceiverReportTime() int64 { r.lock.RLock() defer r.lock.RUnlock() @@ -712,7 +718,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek nowNTP = mediatransportutil.NtpTime(publisherSRData.NtpTimestamp) nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset } else { - timeSincePublisherSRAdjusted := time.Since(time.Unix(0, publisherSRData.AtAdjusted)) + timeSincePublisherSRAdjusted := time.Duration(mono.UnixNano() - publisherSRData.AtAdjusted) reportTimeAdjusted = publisherSRData.AtAdjusted + timeSincePublisherSRAdjusted.Nanoseconds() reportTime = reportTimeAdjusted @@ -732,20 +738,23 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek Octets: octetCount, } - logger := r.logger.WithUnlikelyValues( - "curr", WrappedRTCPSenderReportStateLogger{srData}, - "feed", WrappedRTCPSenderReportStateLogger{publisherSRData}, - "tsOffset", tsOffset, - "timeNow", time.Now(), - "reportTime", time.Unix(0, reportTime), - "reportTimeAdjusted", time.Unix(0, reportTimeAdjusted), - "timeSinceHighest", time.Since(time.Unix(0, r.highestTime)), - "timeSinceFirst", time.Since(time.Unix(0, r.firstTime)), - "timeSincePublisherSRAdjusted", time.Since(time.Unix(0, publisherSRData.AtAdjusted)), - "timeSincePublisherSR", time.Since(time.Unix(0, publisherSRData.At)), - "nowRTPExt", nowRTPExt, - "rtpStats", lockedRTPStatsSenderLogEncoder{r}, - ) + ulgr := func() logger.UnlikelyLogger { + nowNano := mono.UnixNano() + return r.logger.WithUnlikelyValues( + "curr", WrappedRTCPSenderReportStateLogger{srData}, + "feed", WrappedRTCPSenderReportStateLogger{publisherSRData}, + "tsOffset", tsOffset, + "timeNow", mono.Now(), + "reportTime", time.Unix(0, reportTime), + "reportTimeAdjusted", time.Unix(0, reportTimeAdjusted), + "timeSinceHighest", time.Duration(nowNano-r.highestTime), + "timeSinceFirst", time.Duration(nowNano-r.firstTime), + "timeSincePublisherSRAdjusted", time.Duration(nowNano-publisherSRData.AtAdjusted), + "timeSincePublisherSR", time.Duration(nowNano-publisherSRData.At), + "nowRTPExt", nowRTPExt, + "rtpStats", lockedRTPStatsSenderLogEncoder{r}, + ) + } if r.srNewest != nil && nowRTPExt >= r.srNewest.RtpTimestampExt { timeSinceLastReport := nowNTP.Time().Sub(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()) @@ -754,7 +763,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek if timeSinceLastReport.Seconds() > 0.2 && math.Abs(float64(r.params.ClockRate)-windowClockRate) > 0.2*float64(r.params.ClockRate) { r.clockSkewCount++ if (r.clockSkewCount-1)%100 == 0 { - logger.Infow( + ulgr().Infow( "sending sender report, clock skew", "timeSinceLastReport", timeSinceLastReport, "rtpDiffSinceLastReport", rtpDiffSinceLastReport, @@ -768,7 +777,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek if r.srNewest != nil && nowRTPExt < r.srNewest.RtpTimestampExt { // If report being generated is behind the last report, skip it. // Should not happen. - logger.Infow("sending sender report, out-of-order, skipping") + ulgr().Infow("sending sender report, out-of-order, skipping") return nil } @@ -810,7 +819,7 @@ func (r *RTPStatsSender) DeltaInfo(snapshotID uint32) *RTPDeltaInfo { func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo { r.lock.Lock() defer r.lock.Unlock() - if r.lastRRTime.IsZero() { + if r.lastRRTime == 0 { return nil } @@ -830,7 +839,7 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo "senderSnapshotNow", now, "senderSnapshotThen", then, "packetsExpected", packetsExpected, - "duration", endTime.Sub(startTime), + "duration", time.Duration(endTime-startTime), "rtpStats", lockedRTPStatsSenderLogEncoder{r}, ) return nil @@ -857,7 +866,7 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo "packetsExpected", packetsExpected, "packetsLost", packetsLost, "packetsLostFeed", packetsLostFeed, - "duration", endTime.Sub(startTime), + "duration", time.Duration(endTime-startTime), "rtpStats", lockedRTPStatsSenderLogEncoder{r}, ) packetsLost = packetsExpected @@ -871,8 +880,8 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo maxJitterTime := maxJitter / float64(r.params.ClockRate) * 1e6 return &RTPDeltaInfo{ - StartTime: startTime, - EndTime: endTime, + StartTime: time.Unix(0, startTime), + EndTime: time.Unix(0, endTime), Packets: packetsExpected - uint32(now.packetsPadding-then.packetsPadding), Bytes: now.bytes - then.bytes, HeaderBytes: now.headerBytes - then.headerBytes, @@ -928,7 +937,7 @@ func (r *RTPStatsSender) ToProto() *livekit.RTPStats { } func (r *RTPStatsSender) getAndResetSenderSnapshot(senderSnapshotID uint32) (*senderSnapshot, *senderSnapshot) { - if !r.initialized || r.lastRRTime.IsZero() { + if !r.initialized || r.lastRRTime == 0 { return nil, nil } @@ -945,7 +954,7 @@ func (r *RTPStatsSender) getAndResetSenderSnapshot(senderSnapshotID uint32) (*se return &then, &now } -func (r *RTPStatsSender) getSenderSnapshot(startTime time.Time, s *senderSnapshot) senderSnapshot { +func (r *RTPStatsSender) getSenderSnapshot(startTime int64, s *senderSnapshot) senderSnapshot { if s == nil { return senderSnapshot{} } @@ -1113,7 +1122,7 @@ func (r lockedRTPStatsSenderLogEncoder) MarshalLogObject(e zapcore.ObjectEncoder e.AddUint64("extStartTS", r.extStartTS) e.AddUint64("extHighestTS", r.extHighestTS) - e.AddTime("lastRRTime", r.lastRRTime) + e.AddTime("lastRRTime", time.Unix(0, r.lastRRTime)) e.AddReflected("lastRR", r.lastRR) e.AddUint64("extHighestSNFromRR", r.extHighestSNFromRR) e.AddUint64("packetsLostFromRR", r.packetsLostFromRR) @@ -1131,7 +1140,7 @@ func (r lockedRTPStatsSenderLogEncoder) MarshalLogObject(e zapcore.ObjectEncoder // ------------------------------------------------------------------- -func initSenderSnapshot(startTime time.Time, extStartSN uint64) senderSnapshot { +func initSenderSnapshot(startTime int64, extStartSN uint64) senderSnapshot { return senderSnapshot{ isValid: true, startTime: startTime, diff --git a/pkg/sfu/rtpstats/rtpstats_sender_lite.go b/pkg/sfu/rtpstats/rtpstats_sender_lite.go index 9590ae5695e..ae422352465 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender_lite.go +++ b/pkg/sfu/rtpstats/rtpstats_sender_lite.go @@ -15,9 +15,8 @@ package rtpstats import ( - "time" - "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/utils/mono" "go.uber.org/zap/zapcore" ) @@ -38,14 +37,14 @@ func (r *RTPStatsSenderLite) Update(packetTime int64, packetSize int, extSequenc r.lock.Lock() defer r.lock.Unlock() - if !r.endTime.IsZero() { + if r.endTime != 0 { return } if !r.initialized { r.initialized = true - r.startTime = time.Now() + r.startTime = mono.UnixNano() r.extStartSN = extSequenceNumber r.extHighestSN = extSequenceNumber - 1 diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 9724fab98da..c94ed914dc6 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -400,8 +400,6 @@ func (s *StreamAllocator) OnREMB(downTrack *sfu.DownTrack, remb *rtcp.ReceiverEs } // try to lock to track which is sending this update - downTrackSSRC := track.DownTrack().SSRC() - downTrackSSRCRTX := track.DownTrack().SSRCRTX() for _, ssrc := range remb.SSRCs { if ssrc == 0 { continue @@ -1065,6 +1063,12 @@ func (s *StreamAllocator) maybeStopProbe() { func (s *StreamAllocator) maybeBoostDeficientTracks() { availableChannelCapacity := s.getAvailableHeadroom(false) if availableChannelCapacity <= 0 { + s.params.Logger.Infow( + "stream allocator: no available headroom to boost deficient tracks", + "committedChannelCapacity", s.committedChannelCapacity, + "availableChannelCapacity", availableChannelCapacity, + "expectedBandwidthUsage", s.getExpectedBandwidthUsage(), + ) return } diff --git a/pkg/sfu/testutils/data.go b/pkg/sfu/testutils/data.go index 49d5521a9b0..7f7c96872b0 100644 --- a/pkg/sfu/testutils/data.go +++ b/pkg/sfu/testutils/data.go @@ -85,6 +85,9 @@ func GetTestExtPacketVP8(params *TestExtPacketParams, vp8 *buffer.VP8) (*buffer. ep.KeyFrame = vp8.IsKeyFrame ep.Payload = *vp8 + if ep.DependencyDescriptor == nil { + ep.Temporal = int32(vp8.TID) + } return ep, nil } diff --git a/pkg/sfu/utils/mimetype.go b/pkg/sfu/utils/mimetype.go new file mode 100644 index 00000000000..0908e1eec0d --- /dev/null +++ b/pkg/sfu/utils/mimetype.go @@ -0,0 +1,102 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +type MimeType int + +const ( + MimeTypeUnknown MimeType = iota + MimeTypeVP8 + MimeTypeVP9 + MimeTypeH264 + MimeTypeAV1 +) + +func MatchMimeType(mimeType string) MimeType { + switch len(mimeType) { + case 9: + switch mimeType[0] { + case 'v', 'V': + switch mimeType[1] { + case 'i', 'I': + switch mimeType[2] { + case 'd', 'D': + switch mimeType[3] { + case 'e', 'E': + switch mimeType[4] { + case 'o', 'O': + switch mimeType[5] { + case '/': + switch mimeType[6] { + case 'v', 'V': + switch mimeType[7] { + case 'p', 'P': + switch mimeType[8] { + case '8': + return MimeTypeVP8 + case '9': + return MimeTypeVP9 + } + } + case 'a', 'A': + switch mimeType[7] { + case 'v', 'V': + switch mimeType[8] { + case '1': + return MimeTypeAV1 + } + } + } + } + } + } + } + } + } + case 10: + switch mimeType[0] { + case 'v', 'V': + switch mimeType[1] { + case 'i', 'I': + switch mimeType[2] { + case 'd', 'D': + switch mimeType[3] { + case 'e', 'E': + switch mimeType[4] { + case 'o', 'O': + switch mimeType[5] { + case '/': + switch mimeType[6] { + case 'h', 'H': + switch mimeType[7] { + case '2': + switch mimeType[8] { + case '6': + switch mimeType[9] { + case '4': + return MimeTypeH264 + } + } + } + } + } + } + } + } + } + } + } + return MimeTypeUnknown +} diff --git a/pkg/sfu/utils/mimetype_test.go b/pkg/sfu/utils/mimetype_test.go new file mode 100644 index 00000000000..6cfcc7111da --- /dev/null +++ b/pkg/sfu/utils/mimetype_test.go @@ -0,0 +1,65 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func toLowerSwitch(mimeType string) MimeType { + switch strings.ToLower(mimeType) { + case "video/vp8": + return MimeTypeVP8 + case "video/vp9": + return MimeTypeVP9 + case "video/h264": + return MimeTypeH264 + case "video/av1": + return MimeTypeAV1 + default: + return MimeTypeUnknown + } +} + +func TestMimeTypeMatch(t *testing.T) { + require.Equal(t, MimeTypeVP8, MatchMimeType("VIDEO/VP8"), "VIDEO/VP8") + require.Equal(t, MimeTypeVP9, MatchMimeType("VIDEO/VP9"), "VIDEO/VP9") + require.Equal(t, MimeTypeH264, MatchMimeType("VIDEO/H264"), "VIDEO/H264") + require.Equal(t, MimeTypeAV1, MatchMimeType("VIDEO/AV1"), "VIDEO/AV1") +} + +func BenchmarkMimeTypeMatch(b *testing.B) { + mimeTypes := []string{ + "video/VP8", + "video/VP9", + "video/H264", + "video/AV1", + } + + b.Run("ToLower/switch", func(b *testing.B) { + for i := range b.N { + _ = toLowerSwitch(mimeTypes[i%len(mimeTypes)]) + } + }) + + b.Run("MatchMimeType", func(b *testing.B) { + for i := range b.N { + _ = MatchMimeType(mimeTypes[i%len(mimeTypes)]) + } + }) +} diff --git a/pkg/sfu/utils/wraparound.go b/pkg/sfu/utils/wraparound.go index 2a30a63250e..c75db344839 100644 --- a/pkg/sfu/utils/wraparound.go +++ b/pkg/sfu/utils/wraparound.go @@ -120,7 +120,7 @@ func (w *WrapAround[T, ET]) UndoUpdate(result WrapAroundUpdateResult[ET]) { } func (w *WrapAround[T, ET]) Rollover(val T, numCycles int) (result WrapAroundUpdateResult[ET]) { - if !w.initialized || numCycles < 0 { + if numCycles < 0 || !w.initialized { return w.Update(val) } diff --git a/pkg/sfu/videolayerselector/framechain.go b/pkg/sfu/videolayerselector/framechain.go index 533fc4e7d33..19ddf7e2ccc 100644 --- a/pkg/sfu/videolayerselector/framechain.go +++ b/pkg/sfu/videolayerselector/framechain.go @@ -54,7 +54,7 @@ func (fc *FrameChain) OnFrame(extFrameNum uint64, fd *dd.FrameDependencyTemplate if fd.ChainDiffs[fc.chainIdx] == 0 { if fc.broken { fc.broken = false - fc.logger.Debugw("frame chain intact", "chanIdx", fc.chainIdx, "frame", extFrameNum) + // fc.logger.Debugw("frame chain intact", "chanIdx", fc.chainIdx, "frame", extFrameNum) } fc.expectFrames = fc.expectFrames[:0] return true @@ -86,7 +86,7 @@ func (fc *FrameChain) OnFrame(extFrameNum uint64, fd *dd.FrameDependencyTemplate if !intact { fc.broken = true - fc.logger.Debugw("frame chain broken", "chanIdx", fc.chainIdx, "sd", sd, "frame", extFrameNum, "prevFrame", prevFrameInChain) + // fc.logger.Debugw("frame chain broken", "chanIdx", fc.chainIdx, "sd", sd, "frame", extFrameNum, "prevFrame", prevFrameInChain) } return intact } @@ -100,7 +100,7 @@ func (fc *FrameChain) OnExpectFrameChanged(frameNum uint64, decision selectorDec if f == frameNum { if decision != selectorDecisionForwarded { fc.broken = true - fc.logger.Debugw("frame chain broken", "chanIdx", fc.chainIdx, "sd", decision, "frame", frameNum) + // fc.logger.Debugw("frame chain broken", "chanIdx", fc.chainIdx, "sd", decision, "frame", frameNum) } fc.expectFrames[i] = fc.expectFrames[len(fc.expectFrames)-1] fc.expectFrames = fc.expectFrames[:len(fc.expectFrames)-1] diff --git a/pkg/sfu/videolayerselector/temporallayerselector/vp8.go b/pkg/sfu/videolayerselector/temporallayerselector/vp8.go index 2d266089725..c0a86c4f2b7 100644 --- a/pkg/sfu/videolayerselector/temporallayerselector/vp8.go +++ b/pkg/sfu/videolayerselector/temporallayerselector/vp8.go @@ -37,13 +37,13 @@ func (v *VP8) Select(extPkt *buffer.ExtPacket, current int32, target int32) (thi } vp8, ok := extPkt.Payload.(buffer.VP8) - if !ok || !vp8.T { + if !ok { return } - tid := int32(vp8.TID) + tid := extPkt.Temporal if current < target { - if tid > current && tid <= target && vp8.S && vp8.Y { + if tid > current && tid <= target && vp8.S { this = tid next = tid } diff --git a/pkg/telemetry/prometheus/quality.go b/pkg/telemetry/prometheus/quality.go index c9eaf6b65a6..b55a3f8b654 100644 --- a/pkg/telemetry/prometheus/quality.go +++ b/pkg/telemetry/prometheus/quality.go @@ -23,7 +23,6 @@ import ( var ( qualityRating prometheus.Histogram qualityScore prometheus.Histogram - qualityDrop *prometheus.CounterVec ) func initQualityStats(nodeID string, nodeType livekit.NodeType) { @@ -41,21 +40,12 @@ func initQualityStats(nodeID string, nodeType livekit.NodeType) { ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, Buckets: []float64{1.0, 2.0, 2.5, 3.0, 3.25, 3.5, 3.75, 4.0, 4.25, 4.5}, }) - qualityDrop = prometheus.NewCounterVec(prometheus.CounterOpts{ - Namespace: livekitNamespace, - Subsystem: "quality", - Name: "drop", - ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, - }, []string{"direction"}) prometheus.MustRegister(qualityRating) prometheus.MustRegister(qualityScore) - prometheus.MustRegister(qualityDrop) } -func RecordQuality(rating livekit.ConnectionQuality, score float32, numUpDrops int, numDownDrops int) { +func RecordQuality(rating livekit.ConnectionQuality, score float32) { qualityRating.Observe(float64(rating)) qualityScore.Observe(float64(score)) - qualityDrop.WithLabelValues("up").Add(float64(numUpDrops)) - qualityDrop.WithLabelValues("down").Add(float64(numDownDrops)) } diff --git a/version/version.go b/version/version.go index f4051a2dd5e..9e7e6b3bad6 100644 --- a/version/version.go +++ b/version/version.go @@ -14,4 +14,4 @@ package version -const Version = "1.8.0" +const Version = "1.8.3"