Merge remote-tracking branch 'origin/master' into raja_1833

This commit is contained in:
boks1971
2023-09-11 08:42:20 +05:30
23 changed files with 2838 additions and 2285 deletions
+1 -1
View File
@@ -51,7 +51,7 @@ jobs:
go mod tidy
- name: Mage Build
uses: magefile/mage-action@v2
uses: magefile/mage-action@v3
with:
version: latest
args: build
+1 -1
View File
@@ -47,7 +47,7 @@ jobs:
run: go mod download
- name: Generate code
uses: magefile/mage-action@v2
uses: magefile/mage-action@v3
with:
version: latest
args: generate
+5 -5
View File
@@ -18,7 +18,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-20230906055425-e81fd5f6fb3f
github.com/livekit/protocol v1.6.2-0.20230828184341-dfb5162c7c86
github.com/livekit/protocol v1.7.2
github.com/livekit/psrpc v0.3.3
github.com/mackerelio/go-osstat v0.2.4
github.com/magefile/mage v1.15.0
@@ -30,11 +30,11 @@ require (
github.com/pion/interceptor v0.1.18
github.com/pion/rtcp v1.2.10
github.com/pion/rtp v1.8.1
github.com/pion/sctp v1.8.7
github.com/pion/sctp v1.8.8
github.com/pion/sdp/v3 v3.0.6
github.com/pion/transport/v2 v2.2.3
github.com/pion/turn/v2 v2.1.3
github.com/pion/webrtc/v3 v3.2.16
github.com/pion/webrtc/v3 v3.2.19
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.16.0
github.com/redis/go-redis/v9 v9.1.0
@@ -84,7 +84,7 @@ require (
github.com/pion/logging v0.2.2 // indirect
github.com/pion/mdns v0.0.8 // indirect
github.com/pion/randutil v0.1.0 // indirect
github.com/pion/srtp/v2 v2.0.16 // indirect
github.com/pion/srtp/v2 v2.0.17 // indirect
github.com/pion/stun v0.6.1 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/prometheus/client_model v0.3.0 // indirect
@@ -102,6 +102,6 @@ require (
golang.org/x/text v0.13.0 // indirect
golang.org/x/tools v0.13.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20230815205213-6bfd019c3878 // indirect
google.golang.org/grpc v1.57.0 // indirect
google.golang.org/grpc v1.58.0 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
)
+10 -29
View File
@@ -125,12 +125,10 @@ github.com/lithammer/shortuuid/v4 v4.0.0 h1:QRbbVkfgNippHOS8PXDkti4NaWeyYfcBTHtw
github.com/lithammer/shortuuid/v4 v4.0.0/go.mod h1:Zs8puNcrvf2rV9rTH51ZLLcj7ZXqQI3lv67aw4KiB1Y=
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkDaKb5iXdynYrzB84ErPPO4LbRASk58=
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
github.com/livekit/mediatransportutil v0.0.0-20230905085142-e1fcf8eae216 h1:gwGhhhx+vUXR1dZqpKBautkx7qJAXvgCdQxgluBiUqc=
github.com/livekit/mediatransportutil v0.0.0-20230905085142-e1fcf8eae216/go.mod h1:+WIOYwiBMive5T81V8B2wdAc2zQNRjNQiJIcPxMTILY=
github.com/livekit/mediatransportutil v0.0.0-20230906055425-e81fd5f6fb3f h1:b4ri7hQESRSzJWzXXcmANG2hJ4HTj5LM01Ekm8lnQmg=
github.com/livekit/mediatransportutil v0.0.0-20230906055425-e81fd5f6fb3f/go.mod h1:+WIOYwiBMive5T81V8B2wdAc2zQNRjNQiJIcPxMTILY=
github.com/livekit/protocol v1.6.2-0.20230828184341-dfb5162c7c86 h1:QEzGhfIOmGdRw17xIldbYzb1MTsYuVfXSqz8FTyfjWQ=
github.com/livekit/protocol v1.6.2-0.20230828184341-dfb5162c7c86/go.mod h1:/JuO+G/btZ5gNwX2+901L6za3UvVO6DHRXHsv8kkLsU=
github.com/livekit/protocol v1.7.2 h1:TPk8rIv5ZZSx1IU5jaGA2W+RdoDlE8dp4CFHE0MKoGo=
github.com/livekit/protocol v1.7.2/go.mod h1:zbh0QPUcLGOeZeIO/VeigwWWbudz4Lv+Px94FnVfQH0=
github.com/livekit/psrpc v0.3.3 h1:+lltbuN39IdaynXhLLxRShgYqYsRMWeeXKzv60oqyWo=
github.com/livekit/psrpc v0.3.3/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U=
github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs=
@@ -189,47 +187,39 @@ github.com/pion/datachannel v1.5.5 h1:10ef4kwdjije+M9d7Xm9im2Y3O6A6ccQb0zcqZcJew
github.com/pion/datachannel v1.5.5/go.mod h1:iMz+lECmfdCMqFRhXhcA/219B0SQlbpoR2V118yimL0=
github.com/pion/dtls/v2 v2.2.7 h1:cSUBsETxepsCSFSxC3mc/aDo14qQLMSL+O6IjG28yV8=
github.com/pion/dtls/v2 v2.2.7/go.mod h1:8WiMkebSHFD0T+dIU+UeBaoV7kDhOW5oDCzZ7WZ/F9s=
github.com/pion/ice/v2 v2.3.10/go.mod h1:hHGCibDfmXGqukayQw979xEctASp2Pe5Oe0iDU8pRus=
github.com/pion/ice/v2 v2.3.11 h1:rZjVmUwyT55cmN8ySMpL7rsS8KYsJERsrxJLLxpKhdw=
github.com/pion/ice/v2 v2.3.11/go.mod h1:hPcLC3kxMa+JGRzMHqQzjoSj3xtE9F+eoncmXLlCL4E=
github.com/pion/interceptor v0.1.17/go.mod h1:SY8kpmfVBvrbUzvj2bsXz7OJt5JvmVNZ+4Kjq7FcwrI=
github.com/pion/interceptor v0.1.18 h1:Hk26334NUQeUcJNR27YHYKT+sWNhhegQ9KFz5Nn6yMQ=
github.com/pion/interceptor v0.1.18/go.mod h1:tpvvF4cPM6NGxFA1DUMbhabzQBxdWMATDGEUYOR9x6I=
github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
github.com/pion/mdns v0.0.7/go.mod h1:4iP2UbeFhLI/vWju/bw6ZfwjJzk0z8DNValjGxR/dD8=
github.com/pion/mdns v0.0.8 h1:HhicWIg7OX5PVilyBO6plhMetInbzkVJAhbdJiAeVaI=
github.com/pion/mdns v0.0.8/go.mod h1:hYE72WX8WDveIhg7fmXgMKivD3Puklk0Ymzog0lSyaI=
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.10 h1:nkr3uj+8Sp97zyItdN60tE/S6vk4al5CPRR6Gejsdjc=
github.com/pion/rtcp v1.2.10/go.mod h1:ztfEwXZNLGyF1oQDttz/ZKIBaeeg/oWbRYqzBM9TL1I=
github.com/pion/rtp v1.7.13/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
github.com/pion/rtp v1.8.0/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU=
github.com/pion/rtp v1.8.1 h1:26OxTc6lKg/qLSGir5agLyj0QKaOv8OP5wps2SFnVNQ=
github.com/pion/rtp v1.8.1/go.mod h1:pBGHaFt/yW7bf1jjWAoUjpSNoDnw98KTMg+jWWvziqU=
github.com/pion/sctp v1.8.5/go.mod h1:SUFFfDpViyKejTAdwD1d/HQsCu+V/40cCs2nZIvC3s0=
github.com/pion/sctp v1.8.7 h1:JnABvFakZueGAn4KU/4PSKg+GWbF6QWbKTWZOSGJjXw=
github.com/pion/sctp v1.8.7/go.mod h1:g1Ul+ARqZq5JEmoFy87Q/4CePtKnTJ1QCL9dBBdN6AU=
github.com/pion/sctp v1.8.8 h1:5EdnnKI4gpyR1a1TwbiS/wxEgcUWBHsc7ILAjARJB+U=
github.com/pion/sctp v1.8.8/go.mod h1:igF9nZBrjh5AtmKc7U30jXltsFHicFCXSmWA2GWRaWs=
github.com/pion/sdp/v3 v3.0.6 h1:WuDLhtuFUUVpTfus9ILC4HRyHsW6TdugjEX/QY9OiUw=
github.com/pion/sdp/v3 v3.0.6/go.mod h1:iiFWFpQO8Fy3S5ldclBkpXqmWy02ns78NOKoLLL0YQw=
github.com/pion/srtp/v2 v2.0.16 h1:impT2XBrHKsDpXr1x5hHIRydwssrSWKpmw3KvSfXbso=
github.com/pion/srtp/v2 v2.0.16/go.mod h1:NCLCV+U+NpxQ+vXhfOETet4OgKioIgrFjZmIM3ldJYE=
github.com/pion/srtp/v2 v2.0.17 h1:ECuOk+7uIpY6HUlTb0nXhfvu4REG2hjtC4ronYFCZE4=
github.com/pion/srtp/v2 v2.0.17/go.mod h1:y5WSHcJY4YfNB/5r7ca5YjHeIr1H3LM1rKArGGs8jMc=
github.com/pion/stun v0.6.1 h1:8lp6YejULeHBF8NmV8e2787BogQhduZugh5PdhDyyN4=
github.com/pion/stun v0.6.1/go.mod h1:/hO7APkX4hZKu/D0f2lHzNyvdkTGtIy3NDmLR7kSz/8=
github.com/pion/transport v0.14.1 h1:XSM6olwW+o8J4SCmOBb/BpwZypkHeyM0PGFCxNQBr40=
github.com/pion/transport v0.14.1/go.mod h1:4tGmbk00NeYA3rUa9+n+dzCCoKkcy3YlYb99Jn2fNnI=
github.com/pion/transport/v2 v2.0.0/go.mod h1:HS2MEBJTwD+1ZI2eSXSvHJx/HnzQqRy2/LXxt6eVMHc=
github.com/pion/transport/v2 v2.1.0/go.mod h1:AdSw4YBZVDkZm8fpoz+fclXyQwANWmZAlDuQdctTThQ=
github.com/pion/transport/v2 v2.2.0/go.mod h1:AdSw4YBZVDkZm8fpoz+fclXyQwANWmZAlDuQdctTThQ=
github.com/pion/transport/v2 v2.2.1/go.mod h1:cXXWavvCnFF6McHTft3DWS9iic2Mftcz1Aq29pGcU5g=
github.com/pion/transport/v2 v2.2.2/go.mod h1:OJg3ojoBJopjEeECq2yJdXH9YVrUJ1uQ++NjXLOUorc=
github.com/pion/transport/v2 v2.2.3 h1:XcOE3/x41HOSKbl1BfyY1TF1dERx7lVvlMCbXU7kfvA=
github.com/pion/transport/v2 v2.2.3/go.mod h1:q2U/tf9FEfnSBGSW6w5Qp5PFWRLRj3NjLhCCgpRK4p0=
github.com/pion/turn/v2 v2.1.3 h1:pYxTVWG2gpC97opdRc5IGsQ1lJ9O/IlNhkzj7MMrGAA=
github.com/pion/turn/v2 v2.1.3/go.mod h1:huEpByKKHix2/b9kmTAM3YoX6MKP+/D//0ClgUYR2fY=
github.com/pion/webrtc/v3 v3.2.16 h1:2tfQ8qdyUAjeG5Zn44yE98umMtdxuHembJ3WYhj4Zd4=
github.com/pion/webrtc/v3 v3.2.16/go.mod h1:vm5dipobPQGXn2hNyQ+hh2KbTTTaDxJiDcM+MyAyrsc=
github.com/pion/webrtc/v3 v3.2.19 h1:XNu5e62mkzafw1qYuKtQ+Dviw4JpbzC/SLx3zZt49JY=
github.com/pion/webrtc/v3 v3.2.19/go.mod h1:vVURQTBOG5BpWKOJz3nlr23NfTDeyKVmubRNqzQp+Tg=
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=
@@ -262,7 +252,6 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
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.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
@@ -329,9 +318,7 @@ golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b/go.mod h1:9nx3DQGgdP8bBQD5qx
golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
golang.org/x/net v0.5.0/go.mod h1:DivGGAXEgPSlEBzxGzZI+ZLohi+xUj054jfeKui00ws=
golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
golang.org/x/net v0.11.0/go.mod h1:2L/ixqYpgIVXmeoSA/4Lu7BzTG4KIyPIryS4IsOd1oQ=
@@ -382,9 +369,7 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -395,9 +380,7 @@ golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
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.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.4.0/go.mod h1:9P2UbLfCdcvo3p/nzKvsmas4TnlujnuoV9hGgYzW1lQ=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
golang.org/x/term v0.6.0/go.mod h1:m6U89DPEgQRMq3DNkDClhWw02AUbt2daBVO4cn4Hv9U=
golang.org/x/term v0.7.0/go.mod h1:P32HKFT3hSsZrRxla30E9HqToFYAQPCMs/zFMBUFqPY=
golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
golang.org/x/term v0.9.0/go.mod h1:M6DEAAIenWoTxdKrOltXcmDY3rSplQUkrvaDU5FcQyo=
@@ -408,9 +391,7 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
golang.org/x/text v0.6.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
golang.org/x/text v0.10.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
@@ -432,8 +413,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto/googleapis/rpc v0.0.0-20230815205213-6bfd019c3878 h1:lv6/DhyiFFGsmzxbsUUTOkN29II+zeWHxvT8Lpdxsv0=
google.golang.org/genproto/googleapis/rpc v0.0.0-20230815205213-6bfd019c3878/go.mod h1:+Bk1OCOj40wS2hwAMA+aCW9ypzm63QTBBHp6lQ3p+9M=
google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw=
google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo=
google.golang.org/grpc v1.58.0 h1:32JY8YpPMSR45K+c3o6b8VL73V+rR8k+DeMIr4vRH8o=
google.golang.org/grpc v1.58.0/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0=
google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
+2 -2
View File
@@ -327,14 +327,14 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
t.MediaTrackSubscriptions.UpdateVideoLayers()
})
buff.OnFinalRtpStats(func(stats *buffer.RTPStats) {
buff.OnFinalRtpStats(func(stats *livekit.RTPStats) {
t.params.Telemetry.TrackPublishRTPStats(
context.Background(),
t.params.ParticipantID,
t.ID(),
mime,
int(layer),
stats.ToProto(),
stats,
)
})
return newCodec
+9
View File
@@ -16,6 +16,7 @@ package rtc
import (
"context"
"fmt"
"io"
"os"
"strconv"
@@ -69,12 +70,20 @@ type downTrackState struct {
downTrack sfu.DownTrackState
}
// ---------------------------------------------------------------
type participantUpdateInfo struct {
version uint32
state livekit.ParticipantInfo_State
updatedAt time.Time
}
func (p participantUpdateInfo) String() string {
return fmt.Sprintf("version: %d, state: %s, updatedAt: %s", p.version, p.state.String(), p.updatedAt.String())
}
// ---------------------------------------------------------------
type ParticipantParams struct {
Identity livekit.ParticipantIdentity
Name livekit.ParticipantName
+1 -1
View File
@@ -95,7 +95,7 @@ func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit.
// this is a message delivered out of order, a more recent version of the message had already been
// sent.
if pi.Version < lastVersion.version {
p.params.Logger.Debugw("skipping outdated participant update", "version", pi.Version, "lastVersion", lastVersion)
p.params.Logger.Debugw("skipping outdated participant update", "otherParticipant", pi.Identity, "otherPID", pi.Sid, "version", pi.Version, "lastVersion", lastVersion)
isValid = false
}
}
+4 -3
View File
@@ -672,9 +672,10 @@ func (r *RoomManager) handleRTCMessage(ctx context.Context, roomName livekit.Roo
case *livekit.RTCNodeMessage_SendData:
pLogger.Debugw("api send data", "size", len(rm.SendData.Data))
up := &livekit.UserPacket{
Payload: rm.SendData.Data,
DestinationSids: rm.SendData.DestinationSids,
Topic: rm.SendData.Topic,
Payload: rm.SendData.Data,
DestinationSids: rm.SendData.DestinationSids,
DestinationIdentities: rm.SendData.DestinationIdentities,
Topic: rm.SendData.Topic,
}
room.SendDataPacket(up, rm.SendData.Kind)
case *livekit.RTCNodeMessage_UpdateRoomMetadata:
+15 -7
View File
@@ -97,7 +97,7 @@ type Buffer struct {
pliThrottle int64
rtpStats *RTPStats
rtpStats *RTPStatsReceiver
rrSnapshotId uint32
deltaStatsSnapshotId uint32
@@ -108,7 +108,7 @@ type Buffer struct {
onRtcpFeedback func([]rtcp.Packet)
onRtcpSenderReport func()
onFpsChanged func()
onFinalRtpStats func(*RTPStats)
onFinalRtpStats func(*livekit.RTPStats)
// logger
logger logger.Logger
@@ -175,7 +175,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
return
}
b.rtpStats = NewRTPStats(RTPStatsParams{
b.rtpStats = NewRTPStatsReceiver(RTPStatsParams{
ClockRate: codec.ClockRate,
Logger: b.logger,
})
@@ -350,7 +350,7 @@ func (b *Buffer) Close() error {
b.rtpStats.Stop()
b.logger.Infow("rtp stats", "direction", "upstream", "stats", b.rtpStats.ToString())
if b.onFinalRtpStats != nil {
b.onFinalRtpStats(b.rtpStats)
b.onFinalRtpStats(b.rtpStats.ToProto())
}
}
@@ -530,7 +530,15 @@ func (b *Buffer) doFpsCalc(ep *ExtPacket) {
}
func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime time.Time) RTPFlowState {
flowState := b.rtpStats.Update(&p.Header, len(p.Payload), int(p.PaddingSize), arrivalTime)
flowState := b.rtpStats.Update(
arrivalTime,
p.Header.SequenceNumber,
p.Header.Timestamp,
p.Header.Marker,
p.Header.MarshalSize(),
len(p.Payload),
int(p.PaddingSize),
)
if b.nacker != nil {
b.nacker.Remove(p.SequenceNumber)
@@ -693,7 +701,7 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport {
return nil
}
return b.rtpStats.SnapshotRtcpReceptionReport(b.mediaSSRC, b.lastFractionLostToReport, b.rrSnapshotId)
return b.rtpStats.GetRtcpReceptionReport(b.mediaSSRC, b.lastFractionLostToReport, b.rrSnapshotId)
}
func (b *Buffer) SetSenderReportData(rtpTime uint32, ntpTime uint64, packetCount uint32) {
@@ -770,7 +778,7 @@ func (b *Buffer) OnRtcpSenderReport(fn func()) {
b.onRtcpSenderReport = fn
}
func (b *Buffer) OnFinalRtpStats(fn func(*RTPStats)) {
func (b *Buffer) OnFinalRtpStats(fn func(*livekit.RTPStats)) {
b.onFinalRtpStats = fn
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+471
View File
@@ -0,0 +1,471 @@
// 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 buffer
import (
"fmt"
"time"
"github.com/pion/rtcp"
"github.com/livekit/livekit-server/pkg/sfu/utils"
"github.com/livekit/protocol/livekit"
)
type RTPFlowState struct {
IsNotHandled bool
HasLoss bool
LossStartInclusive uint64
LossEndExclusive uint64
IsDuplicate bool
IsOutOfOrder bool
ExtSequenceNumber uint64
ExtTimestamp uint64
}
type RTPStatsReceiver struct {
*rtpStatsBase
resyncOnNextPacket bool
shouldDiscountPaddingOnlyDrops bool
sequenceNumber *utils.WrapAround[uint16, uint64]
timestamp *utils.WrapAround[uint32, uint64]
}
func NewRTPStatsReceiver(params RTPStatsParams) *RTPStatsReceiver {
return &RTPStatsReceiver{
rtpStatsBase: newRTPStatsBase(params),
sequenceNumber: utils.NewWrapAround[uint16, uint64](),
timestamp: utils.NewWrapAround[uint32, uint64](),
}
}
func (r *RTPStatsReceiver) NewSnapshotId() uint32 {
r.lock.Lock()
defer r.lock.Unlock()
return r.newSnapshotID(r.sequenceNumber.GetExtendedStart())
}
func (r *RTPStatsReceiver) Update(
packetTime time.Time,
sequenceNumber uint16,
timestamp uint32,
marker bool,
hdrSize int,
payloadSize int,
paddingSize int,
) (flowState RTPFlowState) {
r.lock.Lock()
defer r.lock.Unlock()
if !r.endTime.IsZero() {
flowState.IsNotHandled = true
return
}
if r.resyncOnNextPacket {
r.resyncOnNextPacket = false
r.resync(packetTime, sequenceNumber, timestamp)
}
var resSN utils.WrapAroundUpdateResult[uint64]
var resTS utils.WrapAroundUpdateResult[uint64]
if !r.initialized {
if payloadSize == 0 {
// do not start on a padding only packet
flowState.IsNotHandled = true
return
}
r.initialized = true
r.startTime = time.Now()
r.firstTime = packetTime
r.highestTime = packetTime
resSN = r.sequenceNumber.Update(sequenceNumber)
resTS = r.timestamp.Update(timestamp)
// initialize snapshots if any
for i := uint32(cFirstSnapshotID); i < r.nextSnapshotID; i++ {
r.snapshots[i] = &snapshot{
startTime: r.startTime,
extStartSN: r.sequenceNumber.GetExtendedStart(),
}
}
r.logger.Debugw(
"rtp receiver stream start",
"startTime", r.startTime.String(),
"firstTime", r.firstTime.String(),
"startSN", r.sequenceNumber.GetExtendedStart(),
"startTS", r.timestamp.GetExtendedStart(),
)
} else {
resSN = r.sequenceNumber.Update(sequenceNumber)
resTS = r.timestamp.Update(timestamp)
}
pktSize := uint64(hdrSize + payloadSize + paddingSize)
gapSN := int64(resSN.ExtendedVal - resSN.PreExtendedHighest)
if gapSN <= 0 { // duplicate OR out-of-order
if payloadSize == 0 {
// do not start on a padding only packet
if resTS.IsRestart {
r.logger.Infow("rolling back timestamp restart", "tsBefore", r.timestamp.GetExtendedStart(), "tsAfter", resTS.PreExtendedStart)
r.timestamp.RollbackRestart(resTS.PreExtendedStart)
}
if resSN.IsRestart {
r.logger.Infow("rolling back sequence number restart", "snBefore", r.sequenceNumber.GetExtendedStart(), "snAfter", resSN.PreExtendedStart)
r.sequenceNumber.RollbackRestart(resSN.PreExtendedStart)
return
}
}
if gapSN != 0 {
r.packetsOutOfOrder++
}
if resSN.IsRestart {
r.packetsLost += resSN.PreExtendedStart - resSN.ExtendedVal
extStartSN := r.sequenceNumber.GetExtendedStart()
for _, s := range r.snapshots {
if s.extStartSN == resSN.PreExtendedStart {
s.extStartSN = extStartSN
}
}
r.logger.Infow(
"adjusting start sequence number",
"snBefore", resSN.PreExtendedStart,
"snAfter", resSN.ExtendedVal,
)
}
if resTS.IsRestart {
r.logger.Infow(
"adjusting start timestamp",
"tsBefore", resTS.PreExtendedStart,
"tsAfter", resTS.ExtendedVal,
)
}
if !r.isSnInfoLost(resSN.ExtendedVal, resSN.PreExtendedHighest) {
r.bytesDuplicate += pktSize
r.headerBytesDuplicate += uint64(hdrSize)
r.packetsDuplicate++
flowState.IsDuplicate = true
} else {
r.packetsLost--
r.setSnInfo(resSN.ExtendedVal, resSN.PreExtendedHighest, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), marker, true)
}
flowState.IsOutOfOrder = true
flowState.ExtSequenceNumber = resSN.ExtendedVal
flowState.ExtTimestamp = resTS.ExtendedVal
} else { // in-order
// update gap histogram
r.updateGapHistogram(int(gapSN))
// update missing sequence numbers
r.clearSnInfos(resSN.PreExtendedHighest+1, resSN.ExtendedVal)
r.packetsLost += uint64(gapSN - 1)
r.setSnInfo(resSN.ExtendedVal, resSN.PreExtendedHighest, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), marker, false)
if timestamp != uint32(resTS.PreExtendedHighest) {
// update only on first packet as same timestamp could be in multiple packets.
// NOTE: this may not be the first packet with this time stamp if there is packet loss.
r.highestTime = packetTime
}
if gapSN > 1 {
flowState.HasLoss = true
flowState.LossStartInclusive = resSN.PreExtendedHighest + 1
flowState.LossEndExclusive = resSN.ExtendedVal
}
flowState.ExtSequenceNumber = resSN.ExtendedVal
flowState.ExtTimestamp = resTS.ExtendedVal
}
if !flowState.IsDuplicate {
if payloadSize == 0 {
r.packetsPadding++
r.bytesPadding += pktSize
r.headerBytesPadding += uint64(hdrSize)
} else {
r.bytes += pktSize
r.headerBytes += uint64(hdrSize)
if marker {
r.frames++
}
r.updateJitter(resTS.ExtendedVal, packetTime)
}
}
return
}
func (r *RTPStatsReceiver) ResyncOnNextPacket(shouldDiscountPaddingOnlyDrops bool) {
r.lock.Lock()
defer r.lock.Unlock()
r.resyncOnNextPacket = true
r.shouldDiscountPaddingOnlyDrops = shouldDiscountPaddingOnlyDrops
}
func (r *RTPStatsReceiver) resync(packetTime time.Time, sn uint16, ts uint32) {
if !r.initialized {
return
}
extHighestSN := r.sequenceNumber.GetExtendedHighest()
var newestPacketCount uint64
var paddingOnlyDrops uint64
var extExpectedHighestSN uint64
var expectedHighestSN uint16
var snCycles uint64
extHighestTS := r.timestamp.GetExtendedHighest()
var newestTS uint64
var extExpectedHighestTS uint64
var expectedHighestTS uint32
var tsCycles uint64
if r.srNewest != nil {
newestPacketCount = r.srNewest.PacketCountExt
paddingOnlyDrops = r.srNewest.PaddingOnlyDrops
if newestPacketCount != 0 {
extExpectedHighestSN = r.sequenceNumber.GetExtendedStart() + newestPacketCount
if r.shouldDiscountPaddingOnlyDrops {
extExpectedHighestSN -= paddingOnlyDrops
}
expectedHighestSN = uint16(extExpectedHighestSN & 0xFFFF)
snCycles = extExpectedHighestSN & 0xFFFF_FFFF_FFFF_0000
if sn-expectedHighestSN < (1<<15) && sn < expectedHighestSN {
snCycles += (1 << 16)
}
if snCycles != 0 && expectedHighestSN-sn < (1<<15) && expectedHighestSN < sn {
snCycles -= (1 << 16)
}
}
newestTS = r.srNewest.RTPTimestampExt
extExpectedHighestTS = newestTS
expectedHighestTS = uint32(extExpectedHighestTS & 0xFFFF_FFFF)
tsCycles = extExpectedHighestTS & 0xFFFF_FFFF_0000_0000
if ts-expectedHighestTS < (1<<31) && ts < expectedHighestTS {
tsCycles += (1 << 32)
}
if tsCycles != 0 && expectedHighestTS-ts < (1<<31) && expectedHighestTS < ts {
tsCycles -= (1 << 32)
}
}
r.sequenceNumber.ResetHighest(snCycles + uint64(sn) - 1)
r.timestamp.ResetHighest(tsCycles + uint64(ts))
r.highestTime = packetTime
r.logger.Debugw(
"resync",
"newestPacketCount", newestPacketCount,
"paddingOnlyDrops", paddingOnlyDrops,
"extExpectedHighestSN", extExpectedHighestSN,
"expectedHighestSN", expectedHighestSN,
"snCycles", snCycles,
"rtpSN", sn,
"beforeExtHighestSN", extHighestSN,
"afterExtHighestSN", r.sequenceNumber.GetExtendedHighest(),
"newestTS", newestTS,
"extExpectedHighestTS", extExpectedHighestTS,
"expectedHighestTS", expectedHighestTS,
"tsCycles", tsCycles,
"rtpTS", ts,
"beforeExtHighestTS", extHighestTS,
"afterExtHighestTS", r.timestamp.GetExtendedHighest(),
)
}
func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData) {
r.lock.Lock()
defer r.lock.Unlock()
if srData == nil || !r.initialized {
return
}
// prevent against extreme case of anachronous sender reports
if r.srNewest != nil && r.srNewest.NTPTimestamp > srData.NTPTimestamp {
r.logger.Infow(
"received anachronous sender report",
"currentNTP", srData.NTPTimestamp.Time().String(),
"currentRTP", srData.RTPTimestamp,
"currentAt", srData.At.String(),
"lastNTP", r.srNewest.NTPTimestamp.Time().String(),
"lastRTP", r.srNewest.RTPTimestamp,
"lastAt", r.srNewest.At.String(),
)
return
}
tsCycles := uint64(0)
pcCycles := uint64(0)
if r.srNewest != nil {
tsCycles = r.srNewest.RTPTimestampExt & 0xFFFF_FFFF_0000_0000
if (srData.RTPTimestamp-r.srNewest.RTPTimestamp) < (1<<31) && srData.RTPTimestamp < r.srNewest.RTPTimestamp {
tsCycles += (1 << 32)
}
pcCycles = r.srNewest.PacketCountExt & 0xFFFF_FFFF_0000_0000
if (srData.PacketCount-r.srNewest.PacketCount) < (1<<31) && srData.PacketCount < r.srNewest.PacketCount {
pcCycles += (1 << 32)
}
}
srDataCopy := *srData
srDataCopy.RTPTimestampExt = uint64(srDataCopy.RTPTimestamp) + tsCycles
srDataCopy.PacketCountExt = uint64(srDataCopy.PacketCount) + pcCycles
r.maybeAdjustFirstPacketTime(srDataCopy.RTPTimestampExt, r.timestamp.GetExtendedStart())
if r.srNewest != nil && srDataCopy.RTPTimestampExt < r.srNewest.RTPTimestampExt {
// This can happen when a track is replaced with a null and then restored -
// i. e. muting replacing with null and unmute restoring the original track.
// Under such a condition reset the sender reports to start from this point.
// Resetting will ensure sample rate calculations do not go haywire due to negative time.
r.logger.Infow(
"received sender report, out-of-order, resetting",
"prevTSExt", r.srNewest.RTPTimestampExt,
"prevRTP", r.srNewest.RTPTimestamp,
"prevNTP", r.srNewest.NTPTimestamp.Time().String(),
"prevAt", r.srNewest.At.String(),
"currTSExt", srDataCopy.RTPTimestampExt,
"currRTP", srDataCopy.RTPTimestamp,
"currNTP", srDataCopy.NTPTimestamp.Time().String(),
"currentAt", srDataCopy.At.String(),
)
r.srFirst = nil
}
r.srNewest = &srDataCopy
if r.srFirst == nil {
r.srFirst = &srDataCopy
}
}
func (r *RTPStatsReceiver) GetRtcpSenderReportData() (srFirst *RTCPSenderReportData, srNewest *RTCPSenderReportData) {
r.lock.RLock()
defer r.lock.RUnlock()
if r.srFirst != nil {
srFirstCopy := *r.srFirst
srFirst = &srFirstCopy
}
if r.srNewest != nil {
srNewestCopy := *r.srNewest
srNewest = &srNewestCopy
}
return
}
func (r *RTPStatsReceiver) GetRtcpReceptionReport(ssrc uint32, proxyFracLost uint8, snapshotID uint32) *rtcp.ReceptionReport {
r.lock.Lock()
defer r.lock.Unlock()
extHighestSN := r.sequenceNumber.GetExtendedHighest()
then, now := r.getAndResetSnapshot(snapshotID, r.sequenceNumber.GetExtendedStart(), extHighestSN)
if now == nil || then == nil {
return nil
}
packetsExpected := now.extStartSN - then.extStartSN
if packetsExpected > cNumSequenceNumbers {
r.logger.Warnw(
"too many packets expected in receiver report",
fmt.Errorf("start: %d, end: %d, expected: %d", then.extStartSN, now.extStartSN, packetsExpected),
)
return nil
}
if packetsExpected == 0 {
return nil
}
intervalStats := r.getIntervalStats(then.extStartSN, now.extStartSN, extHighestSN)
packetsLost := intervalStats.packetsLost
lossRate := float32(packetsLost) / float32(packetsExpected)
fracLost := uint8(lossRate * 256.0)
if proxyFracLost > fracLost {
fracLost = proxyFracLost
}
lastSR := uint32(0)
dlsr := uint32(0)
if r.srNewest != nil {
lastSR = uint32(r.srNewest.NTPTimestamp >> 16)
if !r.srNewest.At.IsZero() {
delayMS := uint32(time.Since(r.srNewest.At).Milliseconds())
dlsr = (delayMS / 1e3) << 16
dlsr |= (delayMS % 1e3) * 65536 / 1000
}
}
return &rtcp.ReceptionReport{
SSRC: ssrc,
FractionLost: fracLost,
TotalLost: uint32(r.packetsLost),
LastSequenceNumber: uint32(now.extStartSN),
Jitter: uint32(r.jitter),
LastSenderReport: lastSR,
Delay: dlsr,
}
}
func (r *RTPStatsReceiver) DeltaInfo(snapshotID uint32) *RTPDeltaInfo {
r.lock.Lock()
defer r.lock.Unlock()
return r.deltaInfo(snapshotID, r.sequenceNumber.GetExtendedStart(), r.sequenceNumber.GetExtendedHighest())
}
func (r *RTPStatsReceiver) ToString() string {
r.lock.RLock()
defer r.lock.RUnlock()
return r.toString(
r.sequenceNumber.GetExtendedStart(), r.sequenceNumber.GetExtendedHighest(), r.timestamp.GetExtendedStart(), r.timestamp.GetExtendedHighest(),
r.packetsLost,
r.jitter, r.maxJitter,
)
}
func (r *RTPStatsReceiver) ToProto() *livekit.RTPStats {
r.lock.RLock()
defer r.lock.RUnlock()
return r.toProto(
r.sequenceNumber.GetExtendedStart(), r.sequenceNumber.GetExtendedHighest(), r.timestamp.GetExtendedStart(), r.timestamp.GetExtendedHighest(),
r.packetsLost,
r.jitter, r.maxJitter,
)
}
// ----------------------------------
@@ -35,9 +35,9 @@ func getPacket(sn uint16, ts uint32, payloadSize int) *rtp.Packet {
}
}
func TestRTPStats(t *testing.T) {
func Test_RTPStatsReceiver(t *testing.T) {
clockRate := uint32(90000)
r := NewRTPStats(RTPStatsParams{
r := NewRTPStatsReceiver(RTPStatsParams{
ClockRate: clockRate,
Logger: logger.GetLogger(),
})
@@ -59,7 +59,15 @@ func TestRTPStats(t *testing.T) {
timestamp += uint32(now.Sub(lastFrameTime).Seconds() * float64(clockRate))
for i := 0; i < packetsPerFrame; i++ {
packet := getPacket(sequenceNumber, timestamp, packetSize)
r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
if (sequenceNumber % 100) == 0 {
jump := uint16(rand.Float64() * 120.0)
sequenceNumber += jump
@@ -77,9 +85,9 @@ func TestRTPStats(t *testing.T) {
fmt.Printf("%s\n", r.ToString())
}
func TestRTPStats_Update(t *testing.T) {
func Test_RTPStatsReceiver_Update(t *testing.T) {
clockRate := uint32(90000)
r := NewRTPStats(RTPStatsParams{
r := NewRTPStatsReceiver(RTPStatsParams{
ClockRate: clockRate,
Logger: logger.GetLogger(),
})
@@ -87,7 +95,15 @@ func TestRTPStats_Update(t *testing.T) {
sequenceNumber := uint16(rand.Float64() * float64(1<<16))
timestamp := uint32(rand.Float64() * float64(1<<32))
packet := getPacket(sequenceNumber, timestamp, 1000)
flowState := r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
flowState := r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
require.False(t, flowState.HasLoss)
require.True(t, r.initialized)
require.Equal(t, sequenceNumber, r.sequenceNumber.GetHighest())
@@ -99,7 +115,15 @@ func TestRTPStats_Update(t *testing.T) {
sequenceNumber++
timestamp += 3000
packet = getPacket(sequenceNumber, timestamp, 1000)
flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
flowState = r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
require.False(t, flowState.HasLoss)
require.Equal(t, sequenceNumber, r.sequenceNumber.GetHighest())
require.Equal(t, sequenceNumber, uint16(r.sequenceNumber.GetExtendedHighest()))
@@ -108,7 +132,15 @@ func TestRTPStats_Update(t *testing.T) {
// out-of-order
packet = getPacket(sequenceNumber-10, timestamp-30000, 1000)
flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
flowState = r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
require.False(t, flowState.HasLoss)
require.Equal(t, sequenceNumber, r.sequenceNumber.GetHighest())
require.Equal(t, sequenceNumber, uint16(r.sequenceNumber.GetExtendedHighest()))
@@ -119,7 +151,15 @@ func TestRTPStats_Update(t *testing.T) {
// duplicate
packet = getPacket(sequenceNumber-10, timestamp-30000, 1000)
flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
flowState = r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
require.False(t, flowState.HasLoss)
require.Equal(t, sequenceNumber, r.sequenceNumber.GetHighest())
require.Equal(t, sequenceNumber, uint16(r.sequenceNumber.GetExtendedHighest()))
@@ -132,7 +172,15 @@ func TestRTPStats_Update(t *testing.T) {
sequenceNumber += 10
timestamp += 30000
packet = getPacket(sequenceNumber, timestamp, 1000)
flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
flowState = r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
require.True(t, flowState.HasLoss)
require.Equal(t, uint64(sequenceNumber-9), flowState.LossStartInclusive)
require.Equal(t, uint64(sequenceNumber), flowState.LossEndExclusive)
@@ -140,7 +188,15 @@ func TestRTPStats_Update(t *testing.T) {
// out-of-order should decrement number of lost packets
packet = getPacket(sequenceNumber-15, timestamp-45000, 1000)
flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
flowState = r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
require.False(t, flowState.HasLoss)
require.Equal(t, sequenceNumber, r.sequenceNumber.GetHighest())
require.Equal(t, sequenceNumber, uint16(r.sequenceNumber.GetExtendedHighest()))
@@ -149,8 +205,112 @@ func TestRTPStats_Update(t *testing.T) {
require.Equal(t, uint64(3), r.packetsOutOfOrder)
require.Equal(t, uint64(1), r.packetsDuplicate)
require.Equal(t, uint64(16), r.packetsLost)
intervalStats := r.getIntervalStats(r.sequenceNumber.GetExtendedStart(), r.sequenceNumber.GetExtendedHighest()+1)
intervalStats := r.getIntervalStats(
r.sequenceNumber.GetExtendedStart(),
r.sequenceNumber.GetExtendedHighest()+1,
r.sequenceNumber.GetExtendedHighest(),
)
require.Equal(t, uint64(16), intervalStats.packetsLost)
// test sequence number cache
// with a gap
sequenceNumber += 2
timestamp += 6000
packet = getPacket(sequenceNumber, timestamp, 1000)
flowState = r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
require.True(t, flowState.HasLoss)
require.Equal(t, uint64(sequenceNumber-1), flowState.LossStartInclusive)
require.Equal(t, uint64(sequenceNumber), flowState.LossEndExclusive)
require.Equal(t, uint64(17), r.packetsLost)
expectedSnInfo := snInfo{
hdrSize: 12,
pktSize: 1012,
isPaddingOnly: false,
marker: false,
isOutOfOrder: false,
}
require.Equal(t, expectedSnInfo, r.snInfos[sequenceNumber&cSnInfoMask])
// out-of-order
sequenceNumber--
timestamp -= 3000
packet = getPacket(sequenceNumber, timestamp, 999)
flowState = r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
0,
)
require.False(t, flowState.HasLoss)
require.Equal(t, uint64(16), r.packetsLost)
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 1011,
isPaddingOnly: false,
marker: false,
isOutOfOrder: true,
}
require.Equal(t, expectedSnInfo, r.snInfos[sequenceNumber&cSnInfoMask])
// check that last one is still fine
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 1012,
isPaddingOnly: false,
marker: false,
isOutOfOrder: false,
}
require.Equal(t, expectedSnInfo, r.snInfos[(sequenceNumber+1)&cSnInfoMask])
// padding only
sequenceNumber += 2
packet = getPacket(sequenceNumber, timestamp, 0)
flowState = r.Update(
time.Now(),
packet.Header.SequenceNumber,
packet.Header.Timestamp,
packet.Header.Marker,
packet.Header.MarshalSize(),
len(packet.Payload),
25,
)
require.False(t, flowState.HasLoss)
require.Equal(t, uint64(16), r.packetsLost)
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 37,
isPaddingOnly: true,
marker: false,
isOutOfOrder: false,
}
require.Equal(t, expectedSnInfo, r.snInfos[sequenceNumber&cSnInfoMask])
// check that last two are still fine
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 1011,
isPaddingOnly: false,
marker: false,
isOutOfOrder: true,
}
require.Equal(t, expectedSnInfo, r.snInfos[(sequenceNumber-2)&cSnInfoMask])
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 1012,
isPaddingOnly: false,
marker: false,
isOutOfOrder: false,
}
require.Equal(t, expectedSnInfo, r.snInfos[(sequenceNumber-1)&cSnInfoMask])
r.Stop()
}
+618
View File
@@ -0,0 +1,618 @@
// 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 buffer
import (
"errors"
"fmt"
"time"
"github.com/pion/rtcp"
"github.com/livekit/mediatransportutil"
"github.com/livekit/protocol/livekit"
)
type senderSnapshot struct {
snapshot
extStartSNFromRR uint64
packetsLostFromRR uint64
maxJitterFromRR float64
}
type RTPStatsSender struct {
*rtpStatsBase
extStartSN uint64
extHighestSN uint64
extHighestSNFromRR uint64
lastRRTime time.Time
lastRR rtcp.ReceptionReport
extStartTS uint64
extHighestTS uint64
packetsLostFromRR uint64
jitterFromRR float64
maxJitterFromRR float64
nextSenderSnapshotID uint32
senderSnapshots map[uint32]*senderSnapshot
}
func NewRTPStatsSender(params RTPStatsParams) *RTPStatsSender {
return &RTPStatsSender{
rtpStatsBase: newRTPStatsBase(params),
nextSenderSnapshotID: cFirstSnapshotID,
senderSnapshots: make(map[uint32]*senderSnapshot),
}
}
func (r *RTPStatsSender) Seed(from *RTPStatsSender) {
r.lock.Lock()
defer r.lock.Unlock()
if !r.seed(from.rtpStatsBase) {
return
}
r.extStartSN = from.extStartSN
r.extHighestSN = from.extHighestSN
r.extHighestSNFromRR = from.extHighestSNFromRR
r.lastRRTime = from.lastRRTime
r.lastRR = from.lastRR
r.extStartTS = from.extStartTS
r.extHighestTS = from.extHighestTS
r.packetsLostFromRR = from.packetsLostFromRR
r.jitterFromRR = from.jitterFromRR
r.maxJitterFromRR = from.maxJitterFromRR
r.nextSenderSnapshotID = from.nextSenderSnapshotID
for id, ss := range from.senderSnapshots {
ssCopy := *ss
r.senderSnapshots[id] = &ssCopy
}
}
func (r *RTPStatsSender) NewSnapshotId() uint32 {
r.lock.Lock()
defer r.lock.Unlock()
return r.newSnapshotID(r.extStartSN)
}
func (r *RTPStatsSender) NewSenderSnapshotId() uint32 {
r.lock.Lock()
defer r.lock.Unlock()
id := r.nextSenderSnapshotID
if r.initialized {
r.senderSnapshots[id] = &senderSnapshot{
snapshot: snapshot{
startTime: time.Now(),
extStartSN: r.extStartSN,
},
extStartSNFromRR: r.extStartSN,
}
}
return id
}
func (r *RTPStatsSender) Update(
packetTime time.Time,
extSequenceNumber uint64,
extTimestamp uint64,
marker bool,
hdrSize int,
payloadSize int,
paddingSize int,
) {
r.lock.Lock()
defer r.lock.Unlock()
if !r.endTime.IsZero() {
return
}
if !r.initialized {
if payloadSize == 0 {
// do not start on a padding only packet
return
}
r.initialized = true
r.startTime = time.Now()
r.firstTime = packetTime
r.highestTime = packetTime
r.extStartSN = extSequenceNumber
r.extHighestSN = extSequenceNumber
r.extStartTS = extTimestamp
r.extHighestTS = extTimestamp
// initialize snapshots if any
for i := uint32(cFirstSnapshotID); i < r.nextSnapshotID; i++ {
r.snapshots[i] = &snapshot{
startTime: r.startTime,
extStartSN: r.extStartSN,
}
}
for i := uint32(cFirstSnapshotID); i < r.nextSenderSnapshotID; i++ {
r.senderSnapshots[i] = &senderSnapshot{
snapshot: snapshot{
startTime: r.startTime,
extStartSN: r.extStartSN,
},
extStartSNFromRR: r.extStartSN,
}
}
r.logger.Debugw(
"rtp sender stream start",
"startTime", r.startTime.String(),
"firstTime", r.firstTime.String(),
"startSN", r.extStartSN,
"startTS", r.extStartTS,
)
}
pktSize := uint64(hdrSize + payloadSize + paddingSize)
isDuplicate := false
gapSN := int64(extSequenceNumber - r.extHighestSN)
if gapSN <= 0 { // duplicate OR out-of-order
if payloadSize == 0 && extSequenceNumber < r.extStartSN {
// do not start on a padding only packet
return
}
if extSequenceNumber < r.extStartSN {
r.packetsLost += r.extStartSN - extSequenceNumber
// adjust start of snapshots
for _, s := range r.snapshots {
if s.extStartSN == r.extStartSN {
s.extStartSN = extSequenceNumber
}
}
for _, s := range r.senderSnapshots {
if s.extStartSN == r.extStartSN {
s.extStartSN = extSequenceNumber
}
}
r.extStartSN = extSequenceNumber
}
if extTimestamp < r.extStartTS {
r.extStartTS = extTimestamp
}
if gapSN != 0 {
r.packetsOutOfOrder++
}
if !r.isSnInfoLost(extSequenceNumber, r.extHighestSN) {
r.bytesDuplicate += pktSize
r.headerBytesDuplicate += uint64(hdrSize)
r.packetsDuplicate++
isDuplicate = true
} else {
r.packetsLost--
r.setSnInfo(extSequenceNumber, r.extHighestSN, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), marker, true)
}
} else { // in-order
// update gap histogram
r.updateGapHistogram(int(gapSN))
// update missing sequence numbers
r.clearSnInfos(r.extHighestSN+1, extSequenceNumber)
r.packetsLost += uint64(gapSN - 1)
r.setSnInfo(extSequenceNumber, r.extHighestSN, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), marker, false)
if extTimestamp != r.extHighestTS {
// update only on first packet as same timestamp could be in multiple packets.
// NOTE: this may not be the first packet with this time stamp if there is packet loss.
r.highestTime = packetTime
}
r.extHighestSN = extSequenceNumber
r.extHighestTS = extTimestamp
}
if !isDuplicate {
if payloadSize == 0 {
r.packetsPadding++
r.bytesPadding += pktSize
r.headerBytesPadding += uint64(hdrSize)
} else {
r.bytes += pktSize
r.headerBytes += uint64(hdrSize)
if marker {
r.frames++
}
jitter := r.updateJitter(extTimestamp, packetTime)
for _, s := range r.senderSnapshots {
if jitter > s.maxJitter {
s.maxJitter = jitter
}
}
}
}
}
func (r *RTPStatsSender) GetTotalPacketsPrimary() uint64 {
r.lock.RLock()
defer r.lock.RUnlock()
return r.getTotalPacketsPrimary(r.extStartSN, r.extHighestSN)
}
func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt uint32, isRttChanged bool) {
r.lock.Lock()
defer r.lock.Unlock()
if !r.initialized || !r.endTime.IsZero() {
return
}
extHighestSNFromRR := r.extHighestSNFromRR&0xFFFF_FFFF_0000_0000 + uint64(rr.LastSequenceNumber)
if !r.lastRRTime.IsZero() {
if (rr.LastSequenceNumber-r.lastRR.LastSequenceNumber) < (1<<31) && rr.LastSequenceNumber < r.lastRR.LastSequenceNumber {
extHighestSNFromRR += (1 << 32)
}
}
if extHighestSNFromRR < r.extStartSN {
// it is possible that the `LastSequenceNumber` in the receiver report is before the starting
// sequence number when dummy packets are used to trigger Pion's OnTrack path.
return
}
var err error
if r.srNewest != nil {
rtt, err = mediatransportutil.GetRttMs(&rr, r.srNewest.NTPTimestamp, r.srNewest.At)
if err == nil {
isRttChanged = rtt != r.rtt
} else {
if !errors.Is(err, mediatransportutil.ErrRttNotLastSenderReport) && !errors.Is(err, mediatransportutil.ErrRttNoLastSenderReport) {
r.logger.Warnw("error getting rtt", err)
}
}
}
if r.lastRRTime.IsZero() || r.extHighestSNFromRR <= extHighestSNFromRR {
r.extHighestSNFromRR = extHighestSNFromRR
packetsLostFromRR := r.packetsLostFromRR&0xFFFF_FFFF_0000_0000 + uint64(rr.TotalLost)
if (rr.TotalLost-r.lastRR.TotalLost) < (1<<31) && rr.TotalLost < r.lastRR.TotalLost {
packetsLostFromRR += (1 << 32)
}
r.packetsLostFromRR = packetsLostFromRR
if isRttChanged {
r.rtt = rtt
if rtt > r.maxRtt {
r.maxRtt = rtt
}
}
r.jitterFromRR = float64(rr.Jitter)
if r.jitterFromRR > r.maxJitterFromRR {
r.maxJitterFromRR = r.jitterFromRR
}
// update snapshots
for _, s := range r.snapshots {
if isRttChanged && rtt > s.maxRtt {
s.maxRtt = rtt
}
}
for _, s := range r.senderSnapshots {
if isRttChanged && rtt > s.maxRtt {
s.maxRtt = rtt
}
if r.jitterFromRR > s.maxJitterFromRR {
s.maxJitterFromRR = r.jitterFromRR
}
}
r.lastRRTime = time.Now()
r.lastRR = rr
} else {
r.logger.Debugw(
fmt.Sprintf("receiver report potentially out of order, highestSN: existing: %d, received: %d", r.extHighestSNFromRR, rr.LastSequenceNumber),
"lastRRTime", r.lastRRTime,
"lastRR", r.lastRR,
"sinceLastRR", time.Since(r.lastRRTime),
"receivedRR", rr,
)
}
return
}
func (r *RTPStatsSender) LastReceiverReportTime() time.Time {
r.lock.RLock()
defer r.lock.RUnlock()
return r.lastRRTime
}
func (r *RTPStatsSender) MaybeAdjustFirstPacketTime(ets uint64) {
r.lock.Lock()
defer r.lock.Unlock()
r.maybeAdjustFirstPacketTime(ets, r.extStartTS)
}
func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt uint64, err error) {
r.lock.RLock()
defer r.lock.RUnlock()
if !r.initialized {
err = errors.New("uninitilaized")
return
}
timeDiff := at.Sub(r.firstTime)
expectedRTPDiff := timeDiff.Nanoseconds() * int64(r.params.ClockRate) / 1e9
expectedTSExt = r.extStartTS + uint64(expectedRTPDiff)
return
}
func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) *rtcp.SenderReport {
r.lock.Lock()
defer r.lock.Unlock()
if !r.initialized {
return nil
}
// construct current time based on monotonic clock
timeSinceFirst := time.Since(r.firstTime)
now := r.firstTime.Add(timeSinceFirst)
nowNTP := mediatransportutil.ToNtpTime(now)
timeSinceHighest := now.Sub(r.highestTime)
nowRTPExt := r.extHighestTS + uint64(timeSinceHighest.Nanoseconds()*int64(r.params.ClockRate)/1e9)
nowRTPExtUsingTime := nowRTPExt
nowRTP := uint32(nowRTPExt)
// It is possible that publisher is pacing at a slower rate.
// That would make `highestTS` to be lagging the RTP time stamp in the RTCP Sender Report from publisher.
// Check for that using calculated clock rate and use the later time stamp if applicable.
var nowRTPExtUsingRate uint64
if calculatedClockRate != 0 {
nowRTPExtUsingRate = r.extStartTS + uint64(float64(calculatedClockRate)*timeSinceFirst.Seconds())
if nowRTPExtUsingRate > nowRTPExt {
nowRTPExt = nowRTPExtUsingRate
nowRTP = uint32(nowRTPExt)
}
}
if r.srNewest != nil && nowRTPExt < r.srNewest.RTPTimestampExt {
// If report being generated is behind, use the time difference and
// clock rate of codec to produce next report.
//
// Current report could be behind due to the following
// - Publisher pacing
// - Due to above, report from publisher side is ahead of packet timestamps.
// Note that report will map wall clock to timestamp at capture time and happens before the pacer.
// - Pause/Mute followed by resume, some combination of events that could
// result in this module not having calculated clock rate of publisher side.
// - When the above happens, current will be generated using highestTS which could be behind.
// That could end up behind the last report's timestamp in extreme cases
r.logger.Infow(
"sending sender report, out-of-order, repairing",
"prevTSExt", r.srNewest.RTPTimestampExt,
"prevRTP", r.srNewest.RTPTimestamp,
"prevNTP", r.srNewest.NTPTimestamp.Time().String(),
"currTSExt", nowRTPExt,
"currRTP", nowRTP,
"currNTP", nowNTP.Time().String(),
"timeNow", time.Now().String(),
"firstTime", r.firstTime.String(),
"timeSinceFirst", timeSinceFirst,
"highestTime", r.highestTime.String(),
"timeSinceHighest", timeSinceHighest,
"nowRTPExtUsingTime", nowRTPExtUsingTime,
"calculatedClockRate", calculatedClockRate,
"nowRTPExtUsingRate", nowRTPExtUsingRate,
)
ntpDiffSinceLast := nowNTP.Time().Sub(r.srNewest.NTPTimestamp.Time())
nowRTPExt = r.srNewest.RTPTimestampExt + uint64(ntpDiffSinceLast.Seconds()*float64(r.params.ClockRate))
nowRTP = uint32(nowRTPExt)
}
r.srNewest = &RTCPSenderReportData{
NTPTimestamp: nowNTP,
RTPTimestamp: nowRTP,
RTPTimestampExt: nowRTPExt,
At: now,
}
if r.srFirst == nil {
r.srFirst = r.srNewest
}
return &rtcp.SenderReport{
SSRC: ssrc,
NTPTime: uint64(nowNTP),
RTPTime: nowRTP,
PacketCount: uint32(r.getTotalPacketsPrimary(r.extStartSN, r.extHighestSN) + r.packetsDuplicate + r.packetsPadding),
OctetCount: uint32(r.bytes + r.bytesDuplicate + r.bytesPadding),
}
}
func (r *RTPStatsSender) DeltaInfo(snapshotID uint32) *RTPDeltaInfo {
r.lock.Lock()
defer r.lock.Unlock()
return r.deltaInfo(snapshotID, r.extStartSN, r.extHighestSN)
}
func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo {
r.lock.Lock()
defer r.lock.Unlock()
if r.lastRRTime.IsZero() {
return nil
}
then, now := r.getAndResetSenderSnapshot(senderSnapshotID)
if now == nil || then == nil {
return nil
}
startTime := then.startTime
endTime := now.startTime
packetsExpected := now.extStartSNFromRR - then.extStartSNFromRR
if packetsExpected > cNumSequenceNumbers {
r.logger.Warnw(
"too many packets expected in delta (sender)",
fmt.Errorf("start: %d, end: %d, expected: %d", then.extStartSNFromRR, now.extStartSNFromRR, packetsExpected),
)
return nil
}
if packetsExpected == 0 {
// not received RTCP RR (OR) publisher is not producing any data
return nil
}
intervalStats := r.getIntervalStats(then.extStartSNFromRR, now.extStartSNFromRR, r.extHighestSN)
packetsLost := now.packetsLostFromRR - then.packetsLostFromRR
if int32(packetsLost) < 0 {
packetsLost = 0
}
if packetsLost > packetsExpected {
r.logger.Warnw(
"unexpected number of packets lost",
fmt.Errorf(
"start: %d, end: %d, expected: %d, lost: report: %d, interval: %d",
then.extStartSNFromRR,
now.extStartSNFromRR,
packetsExpected,
now.packetsLostFromRR-then.packetsLostFromRR,
intervalStats.packetsLost,
),
)
packetsLost = packetsExpected
}
// discount jitter from publisher side + internal processing
maxJitter := then.maxJitterFromRR - then.maxJitter
if maxJitter < 0.0 {
maxJitter = 0.0
}
maxJitterTime := maxJitter / float64(r.params.ClockRate) * 1e6
return &RTPDeltaInfo{
StartTime: startTime,
Duration: endTime.Sub(startTime),
Packets: uint32(packetsExpected - intervalStats.packetsPadding),
Bytes: intervalStats.bytes,
HeaderBytes: intervalStats.headerBytes,
PacketsDuplicate: uint32(now.packetsDuplicate - then.packetsDuplicate),
BytesDuplicate: now.bytesDuplicate - then.bytesDuplicate,
HeaderBytesDuplicate: now.headerBytesDuplicate - then.headerBytesDuplicate,
PacketsPadding: uint32(intervalStats.packetsPadding),
BytesPadding: intervalStats.bytesPadding,
HeaderBytesPadding: intervalStats.headerBytesPadding,
PacketsLost: uint32(packetsLost),
PacketsMissing: uint32(intervalStats.packetsLost),
PacketsOutOfOrder: uint32(intervalStats.packetsOutOfOrder),
Frames: intervalStats.frames,
RttMax: then.maxRtt,
JitterMax: maxJitterTime,
Nacks: now.nacks - then.nacks,
Plis: now.plis - then.plis,
Firs: now.firs - then.firs,
}
}
func (r *RTPStatsSender) ToString() string {
r.lock.RLock()
defer r.lock.RUnlock()
return r.toString(
r.extStartSN, r.extHighestSN, r.extStartTS, r.extHighestTS,
r.packetsLostFromRR,
r.jitterFromRR, r.maxJitterFromRR,
)
}
func (r *RTPStatsSender) ToProto() *livekit.RTPStats {
r.lock.RLock()
defer r.lock.RUnlock()
return r.toProto(
r.extStartSN, r.extHighestSN, r.extStartTS, r.extHighestTS,
r.packetsLostFromRR,
r.jitterFromRR, r.maxJitterFromRR,
)
}
func (r *RTPStatsSender) getAndResetSenderSnapshot(senderSnapshotID uint32) (*senderSnapshot, *senderSnapshot) {
if !r.initialized || r.lastRRTime.IsZero() {
return nil, nil
}
then := r.senderSnapshots[senderSnapshotID]
if then == nil {
then = &senderSnapshot{
snapshot: snapshot{
startTime: r.startTime,
extStartSN: r.extStartSN,
},
extStartSNFromRR: r.extStartSN,
}
r.senderSnapshots[senderSnapshotID] = then
}
// snapshot now
r.senderSnapshots[senderSnapshotID] = &senderSnapshot{
snapshot: snapshot{
startTime: r.lastRRTime,
extStartSN: r.extHighestSN + 1,
packetsDuplicate: r.packetsDuplicate,
bytesDuplicate: r.bytesDuplicate,
headerBytesDuplicate: r.headerBytesDuplicate,
nacks: r.nacks,
plis: r.plis,
firs: r.firs,
maxJitter: r.jitter,
maxRtt: r.rtt,
},
extStartSNFromRR: r.extHighestSNFromRR + 1,
packetsLostFromRR: r.packetsLostFromRR,
maxJitterFromRR: r.jitterFromRR,
}
// make a copy so that it can be used independently
now := *r.senderSnapshots[senderSnapshotID]
return then, &now
}
// -------------------------------------------------------------------
+26 -18
View File
@@ -33,17 +33,25 @@ const (
noReceiverReportTooLongThreshold = 30 * time.Second
)
type ConnectionStatsReceiverProvider interface {
GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayers
}
type ConnectionStatsSenderProvider interface {
GetDeltaStatsSender() map[uint32]*buffer.StreamStatsWithLayers
GetLastReceiverReportTime() time.Time
GetTotalPacketsSent() uint64
}
type ConnectionStatsParams struct {
UpdateInterval time.Duration
MimeType string
IsFECEnabled bool
IncludeRTT bool
IncludeJitter bool
GetDeltaStats func() map[uint32]*buffer.StreamStatsWithLayers
GetDeltaStatsOverridden func() map[uint32]*buffer.StreamStatsWithLayers
GetLastReceiverReportTime func() time.Time
GetTotalPacketsSent func() uint64
Logger logger.Logger
UpdateInterval time.Duration
MimeType string
IsFECEnabled bool
IncludeRTT bool
IncludeJitter bool
ReceiverProvider ConnectionStatsReceiverProvider
SenderProvider ConnectionStatsSenderProvider
Logger logger.Logger
}
type ConnectionStats struct {
@@ -215,7 +223,7 @@ func (cs *ConnectionStats) updateScoreWithAggregate(agg *buffer.RTPDeltaInfo, at
}
func (cs *ConnectionStats) updateScoreFromReceiverReport(at time.Time) (float32, map[uint32]*buffer.StreamStatsWithLayers) {
if cs.params.GetDeltaStatsOverridden == nil || cs.params.GetLastReceiverReportTime == nil || cs.params.GetTotalPacketsSent == nil {
if cs.params.SenderProvider == nil {
return MinMOS, nil
}
@@ -226,10 +234,10 @@ func (cs *ConnectionStats) updateScoreFromReceiverReport(at time.Time) (float32,
return mos, nil
}
streams := cs.params.GetDeltaStatsOverridden()
streams := cs.params.SenderProvider.GetDeltaStatsSender()
if len(streams) == 0 {
// check for receiver report not received for a while
marker := cs.params.GetLastReceiverReportTime()
marker := cs.params.SenderProvider.GetLastReceiverReportTime()
if marker.IsZero() || streamingStartedAt.After(marker) {
marker = streamingStartedAt
}
@@ -246,7 +254,7 @@ func (cs *ConnectionStats) updateScoreFromReceiverReport(at time.Time) (float32,
// delta stat duration could be large due to not receiving receiver report for a long time (for example, due to mute),
// adjust to streaming start if necessary
agg := toAggregateDeltaInfo(streams)
if streamingStartedAt.After(cs.params.GetLastReceiverReportTime()) {
if streamingStartedAt.After(cs.params.SenderProvider.GetLastReceiverReportTime()) {
// last receiver report was before streaming started, wait for next one
mos, _ := cs.scorer.GetMOSAndQuality()
return mos, streams
@@ -260,16 +268,16 @@ func (cs *ConnectionStats) updateScoreFromReceiverReport(at time.Time) (float32,
}
func (cs *ConnectionStats) updateScoreAt(at time.Time) (float32, map[uint32]*buffer.StreamStatsWithLayers) {
if cs.params.GetDeltaStatsOverridden != nil {
if cs.params.SenderProvider != nil {
// receiver report based quality scoring, use stats from receiver report for scoring
return cs.updateScoreFromReceiverReport(at)
}
if cs.params.GetDeltaStats == nil {
if cs.params.ReceiverProvider == nil {
return MinMOS, nil
}
streams := cs.params.GetDeltaStats()
streams := cs.params.ReceiverProvider.GetDeltaStats()
if len(streams) == 0 {
mos, _ := cs.scorer.GetMOSAndQuality()
return mos, nil
@@ -287,7 +295,7 @@ func (cs *ConnectionStats) updateStreamingStart(at time.Time) time.Time {
cs.lock.Lock()
defer cs.lock.Unlock()
packetsSent := cs.params.GetTotalPacketsSent()
packetsSent := cs.params.SenderProvider.GetTotalPacketsSent()
if packetsSent > cs.packetsSent {
if cs.streamingStartedAt.IsZero() {
// the start could be anywhere after last update, but using `at` as this is not required to be accurate
@@ -31,25 +31,42 @@ func newConnectionStats(
isFECEnabled bool,
includeRTT bool,
includeJitter bool,
getDeltaStats func() map[uint32]*buffer.StreamStatsWithLayers,
receiverProvider ConnectionStatsReceiverProvider,
) *ConnectionStats {
return NewConnectionStats(ConnectionStatsParams{
MimeType: mimeType,
IsFECEnabled: isFECEnabled,
IncludeRTT: includeRTT,
IncludeJitter: includeJitter,
GetDeltaStats: getDeltaStats,
Logger: logger.GetLogger(),
MimeType: mimeType,
IsFECEnabled: isFECEnabled,
IncludeRTT: includeRTT,
IncludeJitter: includeJitter,
ReceiverProvider: receiverProvider,
Logger: logger.GetLogger(),
})
}
// -----------------------------------------------
type testReceiverProvider struct {
streams map[uint32]*buffer.StreamStatsWithLayers
}
func newTestReceiverProvider() *testReceiverProvider {
return &testReceiverProvider{}
}
func (trp *testReceiverProvider) setStreams(streams map[uint32]*buffer.StreamStatsWithLayers) {
trp.streams = streams
}
func (trp *testReceiverProvider) GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayers {
return trp.streams
}
// -----------------------------------------------
func TestConnectionQuality(t *testing.T) {
trp := newTestReceiverProvider()
t.Run("quality scorer operation", func(t *testing.T) {
var streams map[uint32]*buffer.StreamStatsWithLayers
getDeltaStats := func() map[uint32]*buffer.StreamStatsWithLayers {
return streams
}
cs := newConnectionStats("audio/opus", false, true, true, getDeltaStats)
cs := newConnectionStats("audio/opus", false, true, true, trp)
duration := 5 * time.Second
now := time.Now()
@@ -63,7 +80,7 @@ func TestConnectionQuality(t *testing.T) {
require.Equal(t, livekit.ConnectionQuality_EXCELLENT, quality)
// best conditions (no loss, jitter/rtt = 0) - quality should stay EXCELLENT
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -71,7 +88,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 250,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.6), mos)
@@ -79,7 +96,7 @@ func TestConnectionQuality(t *testing.T) {
// introduce loss and the score should drop - 12% loss for Opus -> POOR
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -96,7 +113,7 @@ func TestConnectionQuality(t *testing.T) {
PacketsLost: 0,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(2.1), mos)
@@ -106,7 +123,7 @@ func TestConnectionQuality(t *testing.T) {
// although significant loss (12%) in the previous window, lowest score is
// bound so that climbing back does not take too long even under excellent conditions.
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -114,7 +131,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 250,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.1), mos)
@@ -122,7 +139,7 @@ func TestConnectionQuality(t *testing.T) {
// should stay at GOOD if conditions continue to be good
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -130,7 +147,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 250,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.1), mos)
@@ -138,7 +155,7 @@ func TestConnectionQuality(t *testing.T) {
// should climb up to EXCELLENT if conditions continue to be good
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -146,7 +163,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 250,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.6), mos)
@@ -154,7 +171,7 @@ func TestConnectionQuality(t *testing.T) {
// introduce loss and the score should drop - 5% loss for Opus -> GOOD
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -163,7 +180,7 @@ func TestConnectionQuality(t *testing.T) {
PacketsLost: 13,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.1), mos)
@@ -171,7 +188,7 @@ func TestConnectionQuality(t *testing.T) {
// should stay at GOOD quality for another iteration even if the conditions improve
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -179,7 +196,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 250,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.1), mos)
@@ -187,7 +204,7 @@ func TestConnectionQuality(t *testing.T) {
// should climb up to EXCELLENT if conditions continue to be good
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -195,7 +212,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 250,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.6), mos)
@@ -203,7 +220,7 @@ func TestConnectionQuality(t *testing.T) {
// mute when quality is POOR should return quality to EXCELLENT
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -212,7 +229,7 @@ func TestConnectionQuality(t *testing.T) {
PacketsLost: 30,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(2.1), mos)
@@ -228,7 +245,7 @@ func TestConnectionQuality(t *testing.T) {
// that means even if the next update has 0 packets, it should hold state and stay at EXCELLENT quality
cs.UpdateMuteAt(false, now.Add(3*time.Second))
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -236,7 +253,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 0,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.6), mos)
@@ -244,7 +261,7 @@ func TestConnectionQuality(t *testing.T) {
// next update with no packets should knock quality down
now = now.Add(duration)
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -252,7 +269,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 0,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(2.1), mos)
@@ -265,7 +282,7 @@ func TestConnectionQuality(t *testing.T) {
// with lesser number of packet (simulating DTX).
// even higher loss (like 10%) should not knock down quality due to quadratic weighting of packet loss ratio
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -274,7 +291,7 @@ func TestConnectionQuality(t *testing.T) {
PacketsLost: 5,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.6), mos)
@@ -287,7 +304,7 @@ func TestConnectionQuality(t *testing.T) {
// RTT and jitter can knock quality down.
// at 2% loss, quality should stay at EXCELLENT purely based on loss, but with added RTT/jitter, should drop to GOOD
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -298,7 +315,7 @@ func TestConnectionQuality(t *testing.T) {
JitterMax: 30000,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.1), mos)
@@ -313,7 +330,7 @@ func TestConnectionQuality(t *testing.T) {
cs.AddBitrateTransitionAt(1_000_000, now)
cs.AddBitrateTransitionAt(2_000_000, now.Add(2*time.Second))
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -322,7 +339,7 @@ func TestConnectionQuality(t *testing.T) {
Bytes: 8_000_000 / 8 / 5,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.1), mos)
@@ -332,7 +349,7 @@ func TestConnectionQuality(t *testing.T) {
cs.AddBitrateTransitionAt(1_000_000, now)
cs.AddBitrateTransitionAt(2_000_000, now.Add(2*time.Second))
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -341,7 +358,7 @@ func TestConnectionQuality(t *testing.T) {
Bytes: 8_000_000 / 8 / 5,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.1), mos)
@@ -356,7 +373,7 @@ func TestConnectionQuality(t *testing.T) {
// unmute layer
cs.UpdateLayerMuteAt(false, now.Add(2*time.Second))
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -365,7 +382,7 @@ func TestConnectionQuality(t *testing.T) {
Bytes: 8_000_000 / 8 / 5,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.6), mos)
@@ -383,7 +400,7 @@ func TestConnectionQuality(t *testing.T) {
// although conditions are perfect, climbing back from POOR (because of pause above)
// will only climb to GOOD.
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -392,7 +409,7 @@ func TestConnectionQuality(t *testing.T) {
Bytes: 8_000_000 / 8 / 5,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality = cs.GetScoreAndQuality()
require.Greater(t, float32(4.1), mos)
@@ -400,11 +417,7 @@ func TestConnectionQuality(t *testing.T) {
})
t.Run("quality scorer dependent rtt", func(t *testing.T) {
var streams map[uint32]*buffer.StreamStatsWithLayers
getDeltaStats := func() map[uint32]*buffer.StreamStatsWithLayers {
return streams
}
cs := newConnectionStats("audio/opus", false, false, true, getDeltaStats)
cs := newConnectionStats("audio/opus", false, false, true, trp)
duration := 5 * time.Second
now := time.Now()
@@ -414,7 +427,7 @@ func TestConnectionQuality(t *testing.T) {
// RTT does not knock quality down because it is dependent and hence not taken into account
// at 2% loss, quality should stay at EXCELLENT purely based on loss. With high RTT (700 ms)
// quality should drop to GOOD if RTT were taken into consideration
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -424,7 +437,7 @@ func TestConnectionQuality(t *testing.T) {
RttMax: 700,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality := cs.GetScoreAndQuality()
require.Greater(t, float32(4.6), mos)
@@ -432,11 +445,7 @@ func TestConnectionQuality(t *testing.T) {
})
t.Run("quality scorer dependent jitter", func(t *testing.T) {
var streams map[uint32]*buffer.StreamStatsWithLayers
getDeltaStats := func() map[uint32]*buffer.StreamStatsWithLayers {
return streams
}
cs := newConnectionStats("audio/opus", false, true, false, getDeltaStats)
cs := newConnectionStats("audio/opus", false, true, false, trp)
duration := 5 * time.Second
now := time.Now()
@@ -446,7 +455,7 @@ func TestConnectionQuality(t *testing.T) {
// Jitter does not knock quality down because it is dependent and hence not taken into account
// at 2% loss, quality should stay at EXCELLENT purely based on loss. With high jitter (200 ms)
// quality should drop to GOOD if jitter were taken into consideration
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
1: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -456,7 +465,7 @@ func TestConnectionQuality(t *testing.T) {
JitterMax: 200,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality := cs.GetScoreAndQuality()
require.Greater(t, float32(4.6), mos)
@@ -601,18 +610,14 @@ func TestConnectionQuality(t *testing.T) {
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
var streams map[uint32]*buffer.StreamStatsWithLayers
getDeltaStats := func() map[uint32]*buffer.StreamStatsWithLayers {
return streams
}
cs := newConnectionStats(tc.mimeType, tc.isFECEnabled, true, true, getDeltaStats)
cs := newConnectionStats(tc.mimeType, tc.isFECEnabled, true, true, trp)
duration := 5 * time.Second
now := time.Now()
cs.StartAt(&livekit.TrackInfo{Type: livekit.TrackType_AUDIO}, now.Add(-duration))
for _, eq := range tc.expectedQualities {
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
123: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -621,7 +626,7 @@ func TestConnectionQuality(t *testing.T) {
PacketsLost: uint32(math.Ceil(eq.packetLossPercentage * float64(tc.packetsExpected) / 100.0)),
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality := cs.GetScoreAndQuality()
require.Greater(t, eq.expectedMOS, mos)
@@ -698,11 +703,7 @@ func TestConnectionQuality(t *testing.T) {
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
var streams map[uint32]*buffer.StreamStatsWithLayers
getDeltaStats := func() map[uint32]*buffer.StreamStatsWithLayers {
return streams
}
cs := newConnectionStats("video/vp8", false, true, true, getDeltaStats)
cs := newConnectionStats("video/vp8", false, true, true, trp)
duration := 5 * time.Second
now := time.Now()
@@ -712,7 +713,7 @@ func TestConnectionQuality(t *testing.T) {
cs.AddBitrateTransitionAt(tr.bitrate, now.Add(tr.offset))
}
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
123: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -721,7 +722,7 @@ func TestConnectionQuality(t *testing.T) {
Bytes: tc.bytes,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality := cs.GetScoreAndQuality()
require.Greater(t, tc.expectedMOS, mos)
@@ -789,11 +790,7 @@ func TestConnectionQuality(t *testing.T) {
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
var streams map[uint32]*buffer.StreamStatsWithLayers
getDeltaStats := func() map[uint32]*buffer.StreamStatsWithLayers {
return streams
}
cs := newConnectionStats("video/vp8", false, true, true, getDeltaStats)
cs := newConnectionStats("video/vp8", false, true, true, trp)
duration := 5 * time.Second
now := time.Now()
@@ -803,7 +800,7 @@ func TestConnectionQuality(t *testing.T) {
cs.AddLayerTransitionAt(tr.distance, now.Add(tr.offset))
}
streams = map[uint32]*buffer.StreamStatsWithLayers{
trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{
123: {
RTPStats: &buffer.RTPDeltaInfo{
StartTime: now,
@@ -811,7 +808,7 @@ func TestConnectionQuality(t *testing.T) {
Packets: 200,
},
},
}
})
cs.updateScoreAt(now.Add(duration))
mos, quality := cs.GetScoreAndQuality()
require.Greater(t, tc.expectedMOS, mos)
+91 -66
View File
@@ -126,14 +126,14 @@ var (
// -------------------------------------------------------------------
type DownTrackState struct {
RTPStats *buffer.RTPStats
DeltaStatsOverriddenSnapshotId uint32
ForwarderState ForwarderState
RTPStats *buffer.RTPStatsSender
DeltaStatsSenderSnapshotId uint32
ForwarderState ForwarderState
}
func (d DownTrackState) String() string {
return fmt.Sprintf("DownTrackState{rtpStats: %s, deltaOverridden: %d, forwarder: %s}",
d.RTPStats.ToString(), d.DeltaStatsOverriddenSnapshotId, d.ForwarderState.String())
return fmt.Sprintf("DownTrackState{rtpStats: %s, deltaSender: %d, forwarder: %s}",
d.RTPStats.ToString(), d.DeltaStatsSenderSnapshotId, d.ForwarderState.String())
}
// -------------------------------------------------------------------
@@ -239,7 +239,7 @@ type DownTrack struct {
bindAndConnectedOnce atomic.Bool
writable atomic.Bool
rtpStats *buffer.RTPStats
rtpStats *buffer.RTPStatsSender
totalRepeatedNACKs atomic.Uint32
@@ -247,8 +247,8 @@ type DownTrack struct {
blankFramesGeneration atomic.Uint32
connectionStats *connectionquality.ConnectionStats
deltaStatsOverriddenSnapshotId uint32
connectionStats *connectionquality.ConnectionStats
deltaStatsSenderSnapshotId uint32
isNACKThrottled atomic.Bool
@@ -304,20 +304,17 @@ func NewDownTrack(params DowntrackParams) (*DownTrack, error) {
d.getExpectedRTPTimestamp,
)
d.rtpStats = buffer.NewRTPStats(buffer.RTPStatsParams{
ClockRate: d.codec.ClockRate,
IsReceiverReportDriven: true,
Logger: params.Logger,
d.rtpStats = buffer.NewRTPStatsSender(buffer.RTPStatsParams{
ClockRate: d.codec.ClockRate,
Logger: params.Logger,
})
d.deltaStatsOverriddenSnapshotId = d.rtpStats.NewSnapshotId()
d.deltaStatsSenderSnapshotId = d.rtpStats.NewSenderSnapshotId()
d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
MimeType: codecs[0].MimeType, // LK-TODO have to notify on codec change
IsFECEnabled: strings.EqualFold(codecs[0].MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(codecs[0].SDPFmtpLine), "fec"),
GetDeltaStatsOverridden: d.getDeltaStatsOverridden,
GetLastReceiverReportTime: func() time.Time { return d.rtpStats.LastReceiverReportTime() },
GetTotalPacketsSent: func() uint64 { return d.rtpStats.GetTotalPacketsPrimary() },
Logger: params.Logger.WithValues("direction", "down"),
MimeType: codecs[0].MimeType, // LK-TODO have to notify on codec change
IsFECEnabled: strings.EqualFold(codecs[0].MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(codecs[0].SDPFmtpLine), "fec"),
SenderProvider: d,
Logger: params.Logger.WithValues("direction", "down"),
})
d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
if onStatsUpdate := d.getOnStatsUpdate(); onStatsUpdate != nil {
@@ -763,11 +760,13 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
Metadata: sendPacketMetadata{
layer: layer,
arrival: extPkt.Arrival,
isKeyFrame: extPkt.KeyFrame,
tp: tp,
pool: pool,
layer: layer,
arrival: extPkt.Arrival,
extSequenceNumber: tp.rtp.extSequenceNumber,
extTimestamp: tp.rtp.extTimestamp,
isKeyFrame: extPkt.KeyFrame,
tp: tp,
pool: pool,
},
OnSent: d.packetSent,
})
@@ -854,8 +853,10 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
Metadata: sendPacketMetadata{
isPadding: true,
disableCounter: true,
extSequenceNumber: snts[i].extSequenceNumber,
extTimestamp: snts[i].extTimestamp,
isPadding: true,
disableCounter: true,
},
OnSent: d.packetSent,
})
@@ -1029,16 +1030,16 @@ func (d *DownTrack) MaxLayer() buffer.VideoLayer {
func (d *DownTrack) GetState() DownTrackState {
dts := DownTrackState{
RTPStats: d.rtpStats,
DeltaStatsOverriddenSnapshotId: d.deltaStatsOverriddenSnapshotId,
ForwarderState: d.forwarder.GetState(),
RTPStats: d.rtpStats,
DeltaStatsSenderSnapshotId: d.deltaStatsSenderSnapshotId,
ForwarderState: d.forwarder.GetState(),
}
return dts
}
func (d *DownTrack) SeedState(state DownTrackState) {
d.rtpStats.Seed(state.RTPStats)
d.deltaStatsOverriddenSnapshotId = state.DeltaStatsOverriddenSnapshotId
d.deltaStatsSenderSnapshotId = state.DeltaStatsSenderSnapshotId
d.forwarder.SeedState(state.ForwarderState)
}
@@ -1352,8 +1353,11 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
Metadata: sendPacketMetadata{},
OnSent: d.packetSent,
Metadata: sendPacketMetadata{
extSequenceNumber: snts[i].extSequenceNumber,
extTimestamp: snts[i].extTimestamp,
},
OnSent: d.packetSent,
})
// only the first frame will need frameEndNeeded to close out the
@@ -1579,20 +1583,20 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
nackMisses := uint32(0)
numRepeatedNACKs := uint32(0)
nackInfos := make([]NackInfo, 0, len(filtered))
for _, meta := range d.sequencer.getPacketsMeta(filtered) {
if disallowedLayers[meta.layer] {
for _, epm := range d.sequencer.getExtPacketMetas(filtered) {
if disallowedLayers[epm.layer] {
continue
}
nackAcks++
nackInfos = append(nackInfos, NackInfo{
SequenceNumber: meta.targetSeqNo,
Timestamp: meta.timestamp,
Attempts: meta.nacked,
SequenceNumber: epm.targetSeqNo,
Timestamp: epm.timestamp,
Attempts: epm.nacked,
})
pktBuff := *src
n, err := d.params.Receiver.ReadRTP(pktBuff, uint8(meta.layer), meta.sourceSeqNo)
n, err := d.params.Receiver.ReadRTP(pktBuff, uint8(epm.layer), epm.sourceSeqNo)
if err != nil {
if err == io.EOF {
break
@@ -1601,7 +1605,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
continue
}
if meta.nacked > 1 {
if epm.nacked > 1 {
numRepeatedNACKs++
}
@@ -1610,15 +1614,15 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
d.params.Logger.Errorw("unmarshalling rtp packet failed in retransmit", err)
continue
}
pkt.Header.Marker = meta.marker
pkt.Header.SequenceNumber = meta.targetSeqNo
pkt.Header.Timestamp = meta.timestamp
pkt.Header.Marker = epm.marker
pkt.Header.SequenceNumber = epm.targetSeqNo
pkt.Header.Timestamp = epm.timestamp
pkt.Header.SSRC = d.ssrc
pkt.Header.PayloadType = d.payloadType
var payload []byte
pool := PacketFactory.Get().(*[]byte)
if d.mime == "video/vp8" && len(pkt.Payload) > 0 && len(meta.codecBytes) != 0 {
if d.mime == "video/vp8" && len(pkt.Payload) > 0 && len(epm.codecBytes) != 0 {
var incomingVP8 buffer.VP8
if err = incomingVP8.Unmarshal(pkt.Payload); err != nil {
d.params.Logger.Errorw("unmarshalling VP8 packet err", err)
@@ -1626,7 +1630,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
continue
}
payload = d.translateVP8PacketTo(&pkt, &incomingVP8, meta.codecBytes, pool)
payload = d.translateVP8PacketTo(&pkt, &incomingVP8, epm.codecBytes, pool)
}
if payload == nil {
payload = (*pool)[:len(pkt.Payload)]
@@ -1635,14 +1639,16 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
d.pacer.Enqueue(pacer.Packet{
Header: &pkt.Header,
Extensions: []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: meta.ddBytes}},
Extensions: []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: epm.ddBytes}},
Payload: payload,
AbsSendTimeExtID: uint8(d.absSendTimeExtID),
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
Metadata: sendPacketMetadata{
isRTX: true,
pool: pool,
extSequenceNumber: epm.extSequenceNumber,
extTimestamp: epm.extTimestamp,
isRTX: true,
pool: pool,
},
OnSent: d.packetSent,
})
@@ -1747,8 +1753,16 @@ func (d *DownTrack) deltaStats(ds *buffer.RTPDeltaInfo) map[uint32]*buffer.Strea
return streamStats
}
func (d *DownTrack) getDeltaStatsOverridden() map[uint32]*buffer.StreamStatsWithLayers {
return d.deltaStats(d.rtpStats.DeltaInfoOverridden(d.deltaStatsOverriddenSnapshotId))
func (d *DownTrack) GetDeltaStatsSender() map[uint32]*buffer.StreamStatsWithLayers {
return d.deltaStats(d.rtpStats.DeltaInfoSender(d.deltaStatsSenderSnapshotId))
}
func (d *DownTrack) GetLastReceiverReportTime() time.Time {
return d.rtpStats.LastReceiverReportTime()
}
func (d *DownTrack) GetTotalPacketsSent() uint64 {
return d.rtpStats.GetTotalPacketsPrimary()
}
func (d *DownTrack) GetNackStats() (totalPackets uint32, totalRepeatedNACKs uint32) {
@@ -1781,7 +1795,6 @@ func (d *DownTrack) sendPaddingOnMute() {
// let uptrack have chance to send packet before we send padding
time.Sleep(waitBeforeSendPaddingOnMute)
d.params.Logger.Debugw("sending padding on mute")
if d.kind == webrtc.RTPCodecTypeVideo {
d.sendPaddingOnMuteForVideo()
} else if d.mime == "audio/opus" {
@@ -1796,6 +1809,9 @@ func (d *DownTrack) sendPaddingOnMuteForVideo() {
if d.rtpStats.IsActive() || d.IsClosed() {
return
}
if i == 0 {
d.params.Logger.Debugw("sending padding on mute")
}
d.WritePaddingRTP(20, true, true)
time.Sleep(paddingOnMuteInterval)
}
@@ -1805,10 +1821,15 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
frameRate := uint32(50)
frameDuration := time.Duration(1000/frameRate) * time.Millisecond
numFrames := frameRate * uint32(maxPaddingOnMuteDuration/time.Second)
first := true
for {
if d.rtpStats.IsActive() || d.IsClosed() || numFrames <= 0 {
return
}
if first {
first = false
d.params.Logger.Debugw("sending padding on mute")
}
snts, _, err := d.forwarder.GetSnTsForBlankFrames(frameRate, 1)
if err != nil {
d.params.Logger.Warnw("could not get SN/TS for blank frame", err)
@@ -1839,6 +1860,8 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
TransportWideExtID: uint8(d.transportWideExtID),
WriteStream: d.writeStream,
Metadata: sendPacketMetadata{
extSequenceNumber: snts[i].extSequenceNumber,
extTimestamp: snts[i].extTimestamp,
// although this is using empty frames, mark as padding as these are used to trigger Pion OnTrack only
isPadding: true,
},
@@ -1852,24 +1875,26 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
}
func (d *DownTrack) HandleRTCPSenderReportData(_payloadType webrtc.PayloadType, layer int32, srData *buffer.RTCPSenderReportData) error {
if layer == d.forwarder.GetReferenceLayerSpatial() {
d.rtpStats.MaybeAdjustFirstPacketTime(srData)
if layer == d.forwarder.GetReferenceLayerSpatial() && srData != nil {
d.rtpStats.MaybeAdjustFirstPacketTime(srData.RTPTimestampExt + d.forwarder.GetReferenceTimestampOffset())
}
return nil
}
type sendPacketMetadata struct {
layer int32
arrival time.Time
isKeyFrame bool
isRTX bool
isPadding bool
disableCounter bool
tp *TranslationParams
pool *[]byte
layer int32
arrival time.Time
extSequenceNumber uint64
extTimestamp uint64
isKeyFrame bool
isRTX bool
isPadding bool
disableCounter bool
tp *TranslationParams
pool *[]byte
}
func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int, sendTime time.Time, sendError error) {
func (d *DownTrack) packetSent(md interface{}, marker bool, hdrSize int, payloadSize int, sendTime time.Time, sendError error) {
spmd, ok := md.(sendPacketMetadata)
if !ok {
d.params.Logger.Errorw("invalid send packet metadata", nil)
@@ -1886,7 +1911,7 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int,
if !spmd.disableCounter {
// STREAM-ALLOCATOR-TODO: remove this stream allocator bytes counter once stream allocator changes fully to pull bytes counter
size := uint32(hdr.MarshalSize() + payloadSize)
size := uint32(hdrSize + payloadSize)
d.streamAllocatorBytesCounter.Add(size)
if spmd.isRTX {
d.bytesRetransmitted.Add(size)
@@ -1901,9 +1926,9 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int,
packetTime = sendTime
}
if spmd.isPadding {
d.rtpStats.Update(hdr, 0, payloadSize, packetTime)
d.rtpStats.Update(packetTime, spmd.extSequenceNumber, spmd.extTimestamp, marker, hdrSize, 0, payloadSize)
} else {
d.rtpStats.Update(hdr, payloadSize, 0, packetTime)
d.rtpStats.Update(packetTime, spmd.extSequenceNumber, spmd.extTimestamp, marker, hdrSize, payloadSize, 0)
}
if spmd.isKeyFrame {
@@ -1912,8 +1937,8 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int,
d.params.Logger.Debugw(
"forwarded key frame",
"layer", spmd.layer,
"rtpsn", hdr.SequenceNumber,
"rtpts", hdr.Timestamp,
"rtpsn", spmd.extSequenceNumber,
"rtpts", spmd.extTimestamp,
)
}
+29 -17
View File
@@ -533,6 +533,13 @@ func (f *Forwarder) GetReferenceLayerSpatial() int32 {
return f.referenceLayerSpatial
}
func (f *Forwarder) GetReferenceTimestampOffset() uint64 {
f.lock.RLock()
defer f.lock.RUnlock()
return f.refTSOffset
}
func (f *Forwarder) isDeficientLocked() bool {
return f.lastAllocation.IsDeficient
}
@@ -1517,22 +1524,23 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) (
if err == nil {
extExpectedTS = tsExt
} else {
rtpDiff := uint64(0)
if !f.preStartTime.IsZero() && f.refTSOffset == 0 {
if !f.preStartTime.IsZero() {
timeSinceFirst := time.Since(f.preStartTime)
rtpDiff = uint64(timeSinceFirst.Nanoseconds() * int64(f.codec.ClockRate) / 1e9)
f.refTSOffset = f.extFirstTS + rtpDiff - extRefTS
f.logger.Infow(
"calculating refTSOffset",
"preStartTime", f.preStartTime.String(),
"extFirstTS", f.extFirstTS,
"timeSinceFirst", timeSinceFirst,
"rtpDiff", rtpDiff,
"extRefTS", extRefTS,
"refTSOffset", f.refTSOffset,
)
rtpDiff := uint64(timeSinceFirst.Nanoseconds() * int64(f.codec.ClockRate) / 1e9)
extExpectedTS = f.extFirstTS + rtpDiff
if f.refTSOffset == 0 {
f.refTSOffset = extExpectedTS - extRefTS
f.logger.Infow(
"calculating refTSOffset",
"preStartTime", f.preStartTime.String(),
"extFirstTS", f.extFirstTS,
"timeSinceFirst", timeSinceFirst,
"rtpDiff", rtpDiff,
"extRefTS", extRefTS,
"refTSOffset", f.refTSOffset,
)
}
}
extExpectedTS += rtpDiff
}
}
extRefTS += f.refTSOffset
@@ -1771,17 +1779,21 @@ func (f *Forwarder) maybeStart() {
f.started = true
f.preStartTime = time.Now()
sequenceNumber := uint16(rand.Intn(1<<14)) + uint16(1<<15) // a random number in third quartile of sequence number space
timestamp := uint32(rand.Intn(1<<30)) + uint32(1<<31) // a random number in third quartile of timestamp space
extPkt := &buffer.ExtPacket{
Packet: &rtp.Packet{
Header: rtp.Header{
SequenceNumber: uint16(rand.Intn(1<<14)) + uint16(1<<15), // a random number in third quartile of sequence number space
Timestamp: uint32(rand.Intn(1<<30)) + uint32(1<<31), // a random number in third quartile of timestamp space
SequenceNumber: sequenceNumber,
Timestamp: timestamp,
},
},
ExtSequenceNumber: uint64(sequenceNumber),
ExtTimestamp: uint64(timestamp),
}
f.rtpMunger.SetLastSnTs(extPkt)
f.extFirstTS = uint64(extPkt.Packet.Timestamp)
f.extFirstTS = uint64(timestamp)
f.logger.Debugw(
"starting with dummy forwarding",
"sequenceNumber", extPkt.Packet.SequenceNumber,
+1 -1
View File
@@ -47,7 +47,7 @@ func (b *Base) SendPacket(p *Packet) (int, error) {
var err error
defer func() {
if p.OnSent != nil {
p.OnSent(p.Metadata, p.Header, len(p.Payload), sendingAt, err)
p.OnSent(p.Metadata, p.Header.Marker, p.Header.MarshalSize(), len(p.Payload), sendingAt, err)
}
}()
+1 -1
View File
@@ -34,7 +34,7 @@ type Packet struct {
TransportWideExtID uint8
WriteStream webrtc.TrackLocalWriter
Metadata interface{}
OnSent func(md interface{}, sentHeader *rtp.Header, payloadSize int, sentTime time.Time, sendError error)
OnSent func(md interface{}, marker bool, hdrSize int, payloadSize int, sentTime time.Time, sendError error)
}
type Pacer interface {
+5 -5
View File
@@ -223,10 +223,10 @@ func NewWebRTCReceiver(
})
w.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
MimeType: w.codec.MimeType,
IsFECEnabled: strings.EqualFold(w.codec.MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(w.codec.SDPFmtpLine), "fec"),
GetDeltaStats: w.getDeltaStats,
Logger: w.logger.WithValues("direction", "up"),
MimeType: w.codec.MimeType,
IsFECEnabled: strings.EqualFold(w.codec.MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(w.codec.SDPFmtpLine), "fec"),
ReceiverProvider: w,
Logger: w.logger.WithValues("direction", "up"),
})
w.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
if w.onStatsUpdate != nil {
@@ -595,7 +595,7 @@ func (w *WebRTCReceiver) GetAudioLevel() (float64, bool) {
return 0, false
}
func (w *WebRTCReceiver) getDeltaStats() map[uint32]*buffer.StreamStatsWithLayers {
func (w *WebRTCReceiver) GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayers {
w.bufferMu.RLock()
defer w.bufferMu.RUnlock()
+28 -7
View File
@@ -72,6 +72,12 @@ type packetMeta struct {
ddBytes []byte
}
type extPacketMeta struct {
packetMeta
extSequenceNumber uint64
extTimestamp uint64
}
// Sequencer stores the packet sequence received by the down track
type sequencer struct {
sync.Mutex
@@ -80,6 +86,7 @@ type sequencer struct {
initialized bool
extHighestSN uint64
snOffset uint64
extHighestTS uint64
meta []packetMeta
snRangeMap *utils.RangeMap[uint64, uint64]
rtt uint32
@@ -126,6 +133,7 @@ func (s *sequencer) push(
if !s.initialized {
s.extHighestSN = extModifiedSN - 1
s.extHighestTS = extModifiedTS
s.updateSNOffset()
}
@@ -149,6 +157,10 @@ func (s *sequencer) push(
}
}
if int64(extModifiedTS-s.extHighestTS) >= 0 {
s.extHighestTS = extModifiedTS
}
slot := (extModifiedSN - snOffset) % uint64(s.size)
s.meta[slot] = packetMeta{
sourceSeqNo: uint16(extIncomingSN),
@@ -213,15 +225,16 @@ func (s *sequencer) pushPadding(extStartSNInclusive uint64, extEndSNInclusive ui
s.updateSNOffset()
}
func (s *sequencer) getPacketsMeta(seqNo []uint16) []packetMeta {
func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta {
s.Lock()
defer s.Unlock()
snOffset := uint64(0)
var err error
packetsMeta := make([]packetMeta, 0, len(seqNo))
extPacketMetas := make([]extPacketMeta, 0, len(seqNo))
refTime := s.getRefTime(time.Now())
highestSN := uint16(s.extHighestSN)
highestTS := uint32(s.extHighestTS)
for _, sn := range seqNo {
diff := highestSN - sn
if diff > (1 << 15) {
@@ -258,14 +271,22 @@ func (s *sequencer) getPacketsMeta(seqNo []uint16) []packetMeta {
meta.nacked++
meta.lastNack = refTime
pm := *meta
pm.codecBytes = append([]byte{}, meta.codecBytes...)
pm.ddBytes = append([]byte{}, meta.ddBytes...)
packetsMeta = append(packetsMeta, pm)
extTS := uint64(meta.timestamp) + (s.extHighestTS & 0xFFFF_FFFF_FFFF_0000)
if meta.timestamp > highestTS {
extTS -= (1 << 32)
}
epm := extPacketMeta{
packetMeta: *meta,
extSequenceNumber: extSN,
extTimestamp: extTS,
}
epm.codecBytes = append([]byte{}, meta.codecBytes...)
epm.ddBytes = append([]byte{}, meta.ddBytes...)
extPacketMetas = append(extPacketMetas, epm)
}
}
return packetsMeta
return extPacketMetas
}
func (s *sequencer) getRefTime(at time.Time) uint32 {
+16 -12
View File
@@ -36,41 +36,45 @@ func Test_sequencer(t *testing.T) {
seq.push(time.Now(), 518, 518+uint64(off), 123, true, 2, nil, nil)
req := []uint16{57, 58, 62, 63, 513, 514, 515, 516, 517}
res := seq.getPacketsMeta(req)
res := seq.getExtPacketMetas(req)
// nothing should be returned as not enough time has elapsed since sending packet
require.Equal(t, 0, len(res))
time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
res = seq.getPacketsMeta(req)
res = seq.getExtPacketMetas(req)
require.Equal(t, len(req), len(res))
for i, val := range res {
require.Equal(t, val.targetSeqNo, req[i])
require.Equal(t, val.sourceSeqNo, req[i]-off)
require.Equal(t, val.layer, int8(2))
require.Equal(t, val.extSequenceNumber, uint64(req[i]))
require.Equal(t, val.extTimestamp, uint64(123))
}
res = seq.getPacketsMeta(req)
res = seq.getExtPacketMetas(req)
require.Equal(t, 0, len(res))
time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
res = seq.getPacketsMeta(req)
res = seq.getExtPacketMetas(req)
require.Equal(t, len(req), len(res))
for i, val := range res {
require.Equal(t, val.targetSeqNo, req[i])
require.Equal(t, val.sourceSeqNo, req[i]-off)
require.Equal(t, val.layer, int8(2))
require.Equal(t, val.extSequenceNumber, uint64(req[i]))
require.Equal(t, val.extTimestamp, uint64(123))
}
seq.push(time.Now(), 521, 521+uint64(off), 123, true, 1, nil, nil)
m := seq.getPacketsMeta([]uint16{521 + off})
m := seq.getExtPacketMetas([]uint16{521 + off})
require.Equal(t, 0, len(m))
time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
m = seq.getPacketsMeta([]uint16{521 + off})
m = seq.getExtPacketMetas([]uint16{521 + off})
require.Equal(t, 1, len(m))
seq.push(time.Now(), 505, 505+uint64(off), 123, false, 1, nil, nil)
m = seq.getPacketsMeta([]uint16{505 + off})
m = seq.getExtPacketMetas([]uint16{505 + off})
require.Equal(t, 0, len(m))
time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
m = seq.getPacketsMeta([]uint16{505 + off})
m = seq.getExtPacketMetas([]uint16{505 + off})
require.Equal(t, 1, len(m))
}
@@ -148,7 +152,7 @@ func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) {
}
time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
g := n.getPacketsMeta(tt.args.seqNo)
g := n.getExtPacketMetas(tt.args.seqNo)
var got []uint16
for _, sn := range g {
got = append(got, sn.sourceSeqNo)
@@ -163,7 +167,7 @@ func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) {
}
}
if !reflect.DeepEqual(got, tt.want) {
t.Errorf("getPacketsMeta() = %v, want %v", got, tt.want)
t.Errorf("getExtPacketMetas() = %v, want %v", got, tt.want)
}
})
}
@@ -242,7 +246,7 @@ func Test_sequencer_getNACKSeqNo_no_exclusion(t *testing.T) {
}
time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
g := n.getPacketsMeta(tt.args.seqNo)
g := n.getExtPacketMetas(tt.args.seqNo)
var got []uint16
for _, sn := range g {
got = append(got, sn.sourceSeqNo)
@@ -257,7 +261,7 @@ func Test_sequencer_getNACKSeqNo_no_exclusion(t *testing.T) {
}
}
if !reflect.DeepEqual(got, tt.want) {
t.Errorf("getPacketsMeta() = %v, want %v", got, tt.want)
t.Errorf("getExtPacketMetas() = %v, want %v", got, tt.want)
}
})
}