From 114cd97025f50ebf99009a3b17583f1cdd576807 Mon Sep 17 00:00:00 2001 From: xtaci Date: Sat, 10 Oct 2020 11:52:31 +0800 Subject: [PATCH] upgrade to kcp-go@v5.6.1 --- README.md | 6 +- go.mod | 2 +- go.sum | 11 ++ vendor/github.com/xtaci/kcp-go/v5/.travis.yml | 2 +- vendor/github.com/xtaci/kcp-go/v5/autotune.go | 64 ++++++++ vendor/github.com/xtaci/kcp-go/v5/fec.go | 54 ++++++- vendor/github.com/xtaci/kcp-go/v5/go.mod | 1 + vendor/github.com/xtaci/kcp-go/v5/go.sum | 11 ++ vendor/github.com/xtaci/kcp-go/v5/kcp.go | 4 +- vendor/github.com/xtaci/kcp-go/v5/readloop.go | 13 +- .../xtaci/kcp-go/v5/readloop_linux.go | 11 +- vendor/github.com/xtaci/kcp-go/v5/sess.go | 149 ++++++++---------- vendor/github.com/xtaci/kcp-go/v5/snmp.go | 2 +- vendor/modules.txt | 2 +- 14 files changed, 214 insertions(+), 118 deletions(-) create mode 100644 vendor/github.com/xtaci/kcp-go/v5/autotune.go diff --git a/README.md b/README.md index cf8daf7..7b31f70 100644 --- a/README.md +++ b/README.md @@ -224,8 +224,6 @@ It is able to detect and correct multiple symbol errors. By adding t check symbo ![FED](FEC.png) -Setting parameters of RS-Code with ```-datashard m -parityshard n``` on **BOTH** KCP Client & KCP Server **MUST** be **IDENTICAL**. - #### DSCP Differentiated services or DiffServ is a computer networking architecture that specifies a simple, scalable and coarse-grained mechanism for classifying and managing network traffic and providing quality of service (QoS) on modern IP networks. DiffServ can, for example, be used to provide low-latency to critical network traffic such as voice or streaming media while providing simple best-effort service to non-critical services such as web traffic or file transfers. @@ -357,13 +355,11 @@ Low-level KCP configuration can be altered by using manual mode like above, make ### Identical Parmeters -The parameters below **MUST** be **IDENTICAL** on **BOTH** side: +These parameters **MUST** be **IDENTICAL** on **BOTH** side: 1. -key 1. -crypt 1. -nocomp -1. -datashard -1. -parityshard 1. -smuxver ### References diff --git a/go.mod b/go.mod index c146b20..7f45226 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/google/gopacket v1.1.17 // indirect github.com/pkg/errors v0.9.1 github.com/urfave/cli v1.21.0 - github.com/xtaci/kcp-go/v5 v5.5.17 + github.com/xtaci/kcp-go/v5 v5.6.1 github.com/xtaci/smux v1.5.14 github.com/xtaci/tcpraw v1.2.25 golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de diff --git a/go.sum b/go.sum index 6bc2643..d268c25 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,7 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/coreos/go-iptables v0.4.2 h1:KH0EwId05JwWIfb96gWvkiT2cbuOu8ygqUaB+yPAwIg= github.com/coreos/go-iptables v0.4.2/go.mod h1:/mVI274lEDI2ns62jHCDnCyBF9Iwsmekav8Dbxlm1MU= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4= github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/gopacket v1.1.17 h1:rMrlX2ZY2UbvT+sdz3+6J+pp2z+msCq9MxTU6ymxbBY= @@ -14,6 +15,9 @@ github.com/mmcloughlin/avo v0.0.0-20200803215136-443f81d77104 h1:ULR/QWMgcgRiZLU github.com/mmcloughlin/avo v0.0.0-20200803215136-443f81d77104/go.mod h1:wqKykBG2QzQDJEzvRkcS8x6MiSJkF52hXZsXcjaB3ls= 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/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/templexxx/cpu v0.0.1 h1:hY4WdLOgKdc8y13EYklu9OUTXik80BkxHoWvTO6MQQY= github.com/templexxx/cpu v0.0.1/go.mod h1:w7Tb+7qgcAlIyX4NhLuDKt78AHA5SzPmq0Wj6HiEnnk= github.com/templexxx/cpu v0.0.7 h1:pUEZn8JBy/w5yzdYWgx+0m0xL9uk6j4K91C5kOViAzo= @@ -24,8 +28,14 @@ github.com/tjfoc/gmsm v1.3.2 h1:7JVkAn5bvUJ7HtU08iW6UiD+UTmJTIToHCfeFzkcCxM= github.com/tjfoc/gmsm v1.3.2/go.mod h1:HaUcFuY0auTiaHB9MHFGCPx5IaLhTUd2atbCFBQXn9w= github.com/urfave/cli v1.21.0 h1:wYSSj06510qPIzGSua9ZqsncMmWE3Zr55KBERygyrxE= github.com/urfave/cli v1.21.0/go.mod h1:lxDj6qX9Q6lWQxIrbrT0nwecwUtRnhVZAJjJZrVUZZQ= +github.com/xtaci/kcp-go/v5 v5.5.16-0.20201009084714-dc0fc2364c92 h1:a9X2tFWnJR0JjYSz2n4PdmTGtDHtlnU7VI5bNjUDsyI= +github.com/xtaci/kcp-go/v5 v5.5.16-0.20201009084714-dc0fc2364c92/go.mod h1:W3kVPyNYwZ06p79dNwFWQOVFrdcBpDBsdyvK8moQrYo= +github.com/xtaci/kcp-go/v5 v5.5.16-0.20201009115342-ce66a98f9547 h1:SFJWLwOnjTJCt7JKf2fYH0RWtWJTkvYKIEfLvjJfOqM= +github.com/xtaci/kcp-go/v5 v5.5.16-0.20201009115342-ce66a98f9547/go.mod h1:W3kVPyNYwZ06p79dNwFWQOVFrdcBpDBsdyvK8moQrYo= github.com/xtaci/kcp-go/v5 v5.5.17 h1:bkdaqtER0PMlP05BBHfu6W+71kt/NwbAk93KH7F78Ck= github.com/xtaci/kcp-go/v5 v5.5.17/go.mod h1:pVx3jb4LT5edTmPayc77tIU9nRsjGck8wep5ZV/RBO0= +github.com/xtaci/kcp-go/v5 v5.6.1 h1:Pwn0aoeNSPF9dTS7IgiPXn0HEtaIlVb6y5UKWPsx8bI= +github.com/xtaci/kcp-go/v5 v5.6.1/go.mod h1:W3kVPyNYwZ06p79dNwFWQOVFrdcBpDBsdyvK8moQrYo= github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae h1:J0GxkO96kL4WF+AIT3M4mfUVinOCPgf2uUWYFUzN0sM= github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae/go.mod h1:gXtu8J62kEgmN++bm9BVICuT/e8yiLI2KFobd/TRFsE= github.com/xtaci/smux v1.5.14 h1:1j+zJYDZRv9FHaWqCJfH5RPizIm0fSzJIFbfVn8zsfg= @@ -71,4 +81,5 @@ golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1N golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= diff --git a/vendor/github.com/xtaci/kcp-go/v5/.travis.yml b/vendor/github.com/xtaci/kcp-go/v5/.travis.yml index 4003ab2..0e7c4ac 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/.travis.yml +++ b/vendor/github.com/xtaci/kcp-go/v5/.travis.yml @@ -14,7 +14,7 @@ install: - go get github.com/xtaci/kcp-go script: - - go test -coverprofile=coverage.txt -covermode=atomic -bench . + - go test -coverprofile=coverage.txt -covermode=atomic -bench . -timeout 10m after_success: - bash <(curl -s https://codecov.io/bash) diff --git a/vendor/github.com/xtaci/kcp-go/v5/autotune.go b/vendor/github.com/xtaci/kcp-go/v5/autotune.go new file mode 100644 index 0000000..1f85be3 --- /dev/null +++ b/vendor/github.com/xtaci/kcp-go/v5/autotune.go @@ -0,0 +1,64 @@ +package kcp + +const maxAutoTuneSamples = 258 + +// pulse represents a 0/1 signal with time sequence +type pulse struct { + bit bool // 0 or 1 + seq uint32 // sequence of the signal +} + +// autoTune object +type autoTune struct { + pulses [maxAutoTuneSamples]pulse +} + +// Sample adds a signal sample to the pulse buffer +func (tune *autoTune) Sample(bit bool, seq uint32) { + tune.pulses[seq%maxAutoTuneSamples] = pulse{bit, seq} +} + +// Find a period for a given signal +// returns -1 if not found +// +// --- ------ +// | | +// |______________| +// Period +// Falling Edge Rising Edge +func (tune *autoTune) FindPeriod(bit bool) int { + // last pulse and initial index setup + lastPulse := tune.pulses[0] + idx := 1 + + // left edge + var leftEdge int + for ; idx < len(tune.pulses); idx++ { + if lastPulse.bit != bit && tune.pulses[idx].bit == bit { // edge found + if lastPulse.seq+1 == tune.pulses[idx].seq { // ensure edge continuity + leftEdge = idx + break + } + } + lastPulse = tune.pulses[idx] + } + + // right edge + var rightEdge int + lastPulse = tune.pulses[leftEdge] + idx = leftEdge + 1 + + for ; idx < len(tune.pulses); idx++ { + if lastPulse.seq+1 == tune.pulses[idx].seq { // ensure pulses in this level monotonic + if lastPulse.bit == bit && tune.pulses[idx].bit != bit { // edge found + rightEdge = idx + break + } + } else { + return -1 + } + lastPulse = tune.pulses[idx] + } + + return rightEdge - leftEdge +} diff --git a/vendor/github.com/xtaci/kcp-go/v5/fec.go b/vendor/github.com/xtaci/kcp-go/v5/fec.go index 97cd40b..0a203ee 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/fec.go +++ b/vendor/github.com/xtaci/kcp-go/v5/fec.go @@ -13,6 +13,7 @@ const ( typeData = 0xf1 typeParity = 0xf2 fecExpire = 60000 + rxFECMulti = 3 // FEC keeps rxFECMulti* (dataShard+parityShard) ordered packets in memory ) // fecPacket is a decoded FEC packet @@ -45,21 +46,21 @@ type fecDecoder struct { // RS decoder codec reedsolomon.Encoder + + // auto tune fec parameter + autoTune autoTune } -func newFECDecoder(rxlimit, dataShards, parityShards int) *fecDecoder { +func newFECDecoder(dataShards, parityShards int) *fecDecoder { if dataShards <= 0 || parityShards <= 0 { return nil } - if rxlimit < dataShards+parityShards { - return nil - } dec := new(fecDecoder) - dec.rxlimit = rxlimit dec.dataShards = dataShards dec.parityShards = parityShards dec.shardSize = dataShards + parityShards + dec.rxlimit = rxFECMulti * dec.shardSize codec, err := reedsolomon.New(dataShards, parityShards) if err != nil { return nil @@ -73,6 +74,49 @@ func newFECDecoder(rxlimit, dataShards, parityShards int) *fecDecoder { // decode a fec packet func (dec *fecDecoder) decode(in fecPacket) (recovered [][]byte) { + // sample to auto FEC tuner + if in.flag() == typeData { + dec.autoTune.Sample(true, in.seqid()) + } else { + dec.autoTune.Sample(false, in.seqid()) + } + + // check if FEC parameters is out of sync + var shouldTune bool + if int(in.seqid())%dec.shardSize < dec.dataShards { + if in.flag() != typeData { // expect typeData + shouldTune = true + } + } else { + if in.flag() != typeParity { + shouldTune = true + } + } + + if shouldTune { + autoDS := dec.autoTune.FindPeriod(true) + autoPS := dec.autoTune.FindPeriod(false) + + // edges found, we can tune parameters now + if autoDS > 0 && autoPS > 0 && autoDS < 256 && autoPS < 256 { + // and make sure it's different + if autoDS != dec.dataShards || autoPS != dec.parityShards { + dec.dataShards = autoDS + dec.parityShards = autoPS + dec.shardSize = autoDS + autoPS + dec.rxlimit = rxFECMulti * dec.shardSize + codec, err := reedsolomon.New(autoDS, autoPS) + if err != nil { + return nil + } + dec.codec = codec + dec.decodeCache = make([][]byte, dec.shardSize) + dec.flagCache = make([]bool, dec.shardSize) + //log.Println("autotune to :", dec.dataShards, dec.parityShards) + } + } + } + // insertion n := len(dec.rx) - 1 insertIdx := 0 diff --git a/vendor/github.com/xtaci/kcp-go/v5/go.mod b/vendor/github.com/xtaci/kcp-go/v5/go.mod index f80b2bf..ff51020 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/go.mod +++ b/vendor/github.com/xtaci/kcp-go/v5/go.mod @@ -5,6 +5,7 @@ require ( github.com/klauspost/reedsolomon v1.9.9 github.com/mmcloughlin/avo v0.0.0-20200803215136-443f81d77104 // indirect github.com/pkg/errors v0.9.1 + github.com/stretchr/testify v1.6.1 github.com/templexxx/cpu v0.0.7 // indirect github.com/templexxx/xorsimd v0.4.1 github.com/tjfoc/gmsm v1.3.2 diff --git a/vendor/github.com/xtaci/kcp-go/v5/go.sum b/vendor/github.com/xtaci/kcp-go/v5/go.sum index ebb0cc1..db81c71 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/go.sum +++ b/vendor/github.com/xtaci/kcp-go/v5/go.sum @@ -1,3 +1,5 @@ +github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/klauspost/cpuid v1.2.4/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek= github.com/klauspost/cpuid v1.3.1 h1:5JNjFYYQrZeKRJ0734q51WCEEn2huer72Dc7K+R/b6s= github.com/klauspost/cpuid v1.3.1/go.mod h1:bYW4mA6ZgKPob1/Dlai2LviZJO7KGI3uoWLd42rAQw4= @@ -7,6 +9,11 @@ github.com/mmcloughlin/avo v0.0.0-20200803215136-443f81d77104 h1:ULR/QWMgcgRiZLU github.com/mmcloughlin/avo v0.0.0-20200803215136-443f81d77104/go.mod h1:wqKykBG2QzQDJEzvRkcS8x6MiSJkF52hXZsXcjaB3ls= 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= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0= +github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/templexxx/cpu v0.0.1 h1:hY4WdLOgKdc8y13EYklu9OUTXik80BkxHoWvTO6MQQY= github.com/templexxx/cpu v0.0.1/go.mod h1:w7Tb+7qgcAlIyX4NhLuDKt78AHA5SzPmq0Wj6HiEnnk= github.com/templexxx/cpu v0.0.7 h1:pUEZn8JBy/w5yzdYWgx+0m0xL9uk6j4K91C5kOViAzo= @@ -55,4 +62,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IV golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= diff --git a/vendor/github.com/xtaci/kcp-go/v5/kcp.go b/vendor/github.com/xtaci/kcp-go/v5/kcp.go index 3d1984d..0c6c304 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/kcp.go +++ b/vendor/github.com/xtaci/kcp-go/v5/kcp.go @@ -34,8 +34,8 @@ const ( // monotonic reference time point var refTime time.Time = time.Now() -// currentMs returns current elasped monotonic milliseconds since program startup -func currentMs() uint32 { return uint32(time.Now().Sub(refTime) / time.Millisecond) } +// currentMs returns current elapsed monotonic milliseconds since program startup +func currentMs() uint32 { return uint32(time.Since(refTime) / time.Millisecond) } // output_callback is a prototype which ought capture conn and call conn.Write type output_callback func(buf []byte, size int) diff --git a/vendor/github.com/xtaci/kcp-go/v5/readloop.go b/vendor/github.com/xtaci/kcp-go/v5/readloop.go index bc48e84..697395a 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/readloop.go +++ b/vendor/github.com/xtaci/kcp-go/v5/readloop.go @@ -18,12 +18,7 @@ func (s *UDPSession) defaultReadLoop() { atomic.AddUint64(&DefaultSnmp.InErrs, 1) continue } - - if n >= s.headerSize+IKCP_OVERHEAD { - s.packetInput(buf[:n]) - } else { - atomic.AddUint64(&DefaultSnmp.InErrs, 1) - } + s.packetInput(buf[:n]) } else { s.notifyReadError(errors.WithStack(err)) return @@ -35,11 +30,7 @@ func (l *Listener) defaultMonitor() { buf := make([]byte, mtuLimit) for { if n, from, err := l.conn.ReadFrom(buf); err == nil { - if n >= l.headerSize+IKCP_OVERHEAD { - l.packetInput(buf[:n], from) - } else { - atomic.AddUint64(&DefaultSnmp.InErrs, 1) - } + l.packetInput(buf[:n], from) } else { l.notifyReadError(errors.WithStack(err)) return diff --git a/vendor/github.com/xtaci/kcp-go/v5/readloop_linux.go b/vendor/github.com/xtaci/kcp-go/v5/readloop_linux.go index 6226478..be194af 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/readloop_linux.go +++ b/vendor/github.com/xtaci/kcp-go/v5/readloop_linux.go @@ -39,11 +39,6 @@ func (s *UDPSession) readLoop() { continue } - if msg.N < s.headerSize+IKCP_OVERHEAD { - atomic.AddUint64(&DefaultSnmp.InErrs, 1) - continue - } - // source and size has validated s.packetInput(msg.Buffers[0][:msg.N]) } @@ -95,11 +90,7 @@ func (l *Listener) monitor() { if count, err := xconn.ReadBatch(msgs, 0); err == nil { for i := 0; i < count; i++ { msg := &msgs[i] - if msg.N >= l.headerSize+IKCP_OVERHEAD { - l.packetInput(msg.Buffers[0][:msg.N], msg.Addr) - } else { - atomic.AddUint64(&DefaultSnmp.InErrs, 1) - } + l.packetInput(msg.Buffers[0][:msg.N], msg.Addr) } } else { // compatibility issue: diff --git a/vendor/github.com/xtaci/kcp-go/v5/sess.go b/vendor/github.com/xtaci/kcp-go/v5/sess.go index e5ada2b..2dedd74 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/sess.go +++ b/vendor/github.com/xtaci/kcp-go/v5/sess.go @@ -35,9 +35,6 @@ const ( // maximum packet size mtuLimit = 1500 - // FEC keeps rxFECMulti* (dataShard+parityShard) ordered packets in memory - rxFECMulti = 3 - // accept backlog acceptBacklog = 128 ) @@ -155,7 +152,7 @@ func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn } // FEC codec initialization - sess.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards) + sess.fecDecoder = newFECDecoder(dataShards, parityShards) if sess.block != nil { sess.fecEncoder = newFECEncoder(dataShards, parityShards, cryptHeaderSize) } else { @@ -216,7 +213,7 @@ func (s *UDPSession) Read(b []byte) (n int, err error) { return size, nil } - // if necessary resize the stream buffer to guarantee a sufficent buffer space + // if necessary resize the stream buffer to guarantee a sufficient buffer space if cap(s.recvbuf) < size { s.recvbuf = make([]byte, size) } @@ -240,7 +237,7 @@ func (s *UDPSession) Read(b []byte) (n int, err error) { return 0, errors.WithStack(errTimeout) } - delay := s.rd.Sub(time.Now()) + delay := time.Until(s.rd) timeout = time.NewTimer(delay) c = timeout.C } @@ -311,7 +308,7 @@ func (s *UDPSession) WriteBuffers(v [][]byte) (n int, err error) { s.mu.Unlock() return 0, errors.WithStack(errTimeout) } - delay := s.wd.Sub(time.Now()) + delay := time.Until(s.wd) timeout = time.NewTimer(delay) c = timeout.C } @@ -343,7 +340,6 @@ func (s *UDPSession) uncork() { } s.txqueue = s.txqueue[:0] } - return } // Close closes the connection. @@ -655,22 +651,22 @@ func (s *UDPSession) notifyWriteError(err error) { // packet input stage func (s *UDPSession) packetInput(data []byte) { - dataValid := false - if s.block != nil { + decrypted := false + if s.block != nil && len(data) >= cryptHeaderSize { s.block.Decrypt(data, data) data = data[nonceSize:] checksum := crc32.ChecksumIEEE(data[crcSize:]) if checksum == binary.LittleEndian.Uint32(data) { data = data[crcSize:] - dataValid = true + decrypted = true } else { atomic.AddUint64(&DefaultSnmp.InCsumErrors, 1) } } else if s.block == nil { - dataValid = true + decrypted = true } - if dataValid { + if decrypted && len(data) >= IKCP_OVERHEAD { s.kcpInput(data) } } @@ -678,57 +674,58 @@ func (s *UDPSession) packetInput(data []byte) { func (s *UDPSession) kcpInput(data []byte) { var kcpInErrors, fecErrs, fecRecovered, fecParityShards uint64 - if s.fecDecoder != nil { - if len(data) > fecHeaderSize { // must be larger than fec header size + fecFlag := binary.LittleEndian.Uint16(data[4:]) + if fecFlag == typeData || fecFlag == typeParity { // 16bit kcp cmd [81-84] and frg [0-255] will not overlap with FEC type 0x00f1 0x00f2 + if len(data) >= fecHeaderSizePlus2 { f := fecPacket(data) - if f.flag() == typeData || f.flag() == typeParity { // header check - if f.flag() == typeParity { - fecParityShards++ - } + if f.flag() == typeParity { + fecParityShards++ + } - // lock - s.mu.Lock() - recovers := s.fecDecoder.decode(f) - if f.flag() == typeData { - if ret := s.kcp.Input(data[fecHeaderSizePlus2:], true, s.ackNoDelay); ret != 0 { - kcpInErrors++ - } + // lock + s.mu.Lock() + // if fecDecoder is not initialized, create one with default parameter + if s.fecDecoder == nil { + s.fecDecoder = newFECDecoder(1, 1) + } + recovers := s.fecDecoder.decode(f) + if f.flag() == typeData { + if ret := s.kcp.Input(data[fecHeaderSizePlus2:], true, s.ackNoDelay); ret != 0 { + kcpInErrors++ } + } - for _, r := range recovers { - if len(r) >= 2 { // must be larger than 2bytes - sz := binary.LittleEndian.Uint16(r) - if int(sz) <= len(r) && sz >= 2 { - if ret := s.kcp.Input(r[2:sz], false, s.ackNoDelay); ret == 0 { - fecRecovered++ - } else { - kcpInErrors++ - } + for _, r := range recovers { + if len(r) >= 2 { // must be larger than 2bytes + sz := binary.LittleEndian.Uint16(r) + if int(sz) <= len(r) && sz >= 2 { + if ret := s.kcp.Input(r[2:sz], false, s.ackNoDelay); ret == 0 { + fecRecovered++ } else { - fecErrs++ + kcpInErrors++ } } else { fecErrs++ } - // recycle the recovers - xmitBuf.Put(r) + } else { + fecErrs++ } - - // to notify the readers to receive the data - if n := s.kcp.PeekSize(); n > 0 { - s.notifyReadEvent() - } - // to notify the writers - waitsnd := s.kcp.WaitSnd() - if waitsnd < int(s.kcp.snd_wnd) && waitsnd < int(s.kcp.rmt_wnd) { - s.notifyWriteEvent() - } - - s.uncork() - s.mu.Unlock() - } else { - atomic.AddUint64(&DefaultSnmp.InErrs, 1) + // recycle the recovers + xmitBuf.Put(r) } + + // to notify the readers to receive the data + if n := s.kcp.PeekSize(); n > 0 { + s.notifyReadEvent() + } + // to notify the writers + waitsnd := s.kcp.WaitSnd() + if waitsnd < int(s.kcp.snd_wnd) && waitsnd < int(s.kcp.rmt_wnd) { + s.notifyWriteEvent() + } + + s.uncork() + s.mu.Unlock() } else { atomic.AddUint64(&DefaultSnmp.InErrs, 1) } @@ -771,7 +768,6 @@ type ( block BlockCrypt // block encryption dataShards int // FEC data shard parityShards int // FEC parity shard - fecDecoder *fecDecoder // FEC mock initialization conn net.PacketConn // the underlying packet connection ownConn bool // true if we created conn internally, false if provided by caller @@ -779,7 +775,6 @@ type ( sessionLock sync.RWMutex chAccepts chan *UDPSession // Listen() backlog chSessionClosed chan net.Addr // session close queue - headerSize int // the additional header to a KCP frame die chan struct{} // notify the listener has closed dieOnce sync.Once @@ -795,43 +790,45 @@ type ( // packet input stage func (l *Listener) packetInput(data []byte, addr net.Addr) { - dataValid := false - if l.block != nil { + decrypted := false + if l.block != nil && len(data) >= cryptHeaderSize { l.block.Decrypt(data, data) data = data[nonceSize:] checksum := crc32.ChecksumIEEE(data[crcSize:]) if checksum == binary.LittleEndian.Uint32(data) { data = data[crcSize:] - dataValid = true + decrypted = true } else { atomic.AddUint64(&DefaultSnmp.InCsumErrors, 1) } } else if l.block == nil { - dataValid = true + decrypted = true } - if dataValid { + if decrypted && len(data) >= IKCP_OVERHEAD { l.sessionLock.RLock() s, ok := l.sessions[addr.String()] l.sessionLock.RUnlock() var conv, sn uint32 - convValid := false - if l.fecDecoder != nil { - isfec := binary.LittleEndian.Uint16(data[4:]) - if isfec == typeData { + convRecovered := false + fecFlag := binary.LittleEndian.Uint16(data[4:]) + if fecFlag == typeData || fecFlag == typeParity { // 16bit kcp cmd [81-84] and frg [0-255] will not overlap with FEC type 0x00f1 0x00f2 + // packet with FEC + if fecFlag == typeData && len(data) >= fecHeaderSizePlus2+IKCP_OVERHEAD { conv = binary.LittleEndian.Uint32(data[fecHeaderSizePlus2:]) sn = binary.LittleEndian.Uint32(data[fecHeaderSizePlus2+IKCP_SN_OFFSET:]) - convValid = true + convRecovered = true } } else { + // packet without FEC conv = binary.LittleEndian.Uint32(data) sn = binary.LittleEndian.Uint32(data[IKCP_SN_OFFSET:]) - convValid = true + convRecovered = true } if ok { // existing connection - if !convValid || conv == s.kcp.conv { // parity or valid data shard + if !convRecovered || conv == s.kcp.conv { // parity data or valid conversation s.kcpInput(data) } else if sn == 0 { // should replace current connection s.Close() @@ -839,7 +836,7 @@ func (l *Listener) packetInput(data []byte, addr net.Addr) { } } - if s == nil && convValid { // new session + if s == nil && convRecovered { // new session if len(l.chAccepts) < cap(l.chAccepts) { // do not let the new sessions overwhelm accept queue s := newUDPSession(conv, l.dataShards, l.parityShards, l, l.conn, false, addr, l.block) s.kcpInput(data) @@ -917,7 +914,7 @@ func (l *Listener) Accept() (net.Conn, error) { func (l *Listener) AcceptKCP() (*UDPSession, error) { var timeout <-chan time.Time if tdeadline, ok := l.rd.Load().(time.Time); ok && !tdeadline.IsZero() { - timeout = time.After(tdeadline.Sub(time.Now())) + timeout = time.After(time.Until(tdeadline)) } select { @@ -988,7 +985,7 @@ func Listen(laddr string) (net.Listener, error) { return ListenWithOptions(laddr // // 'block' is the block encryption algorithm to encrypt packets. // -// 'dataShards', 'parityShards' specifiy how many parity packets will be generated following the data packets. +// 'dataShards', 'parityShards' specify how many parity packets will be generated following the data packets. // // Check https://github.com/klauspost/reedsolomon for details func ListenWithOptions(laddr string, block BlockCrypt, dataShards, parityShards int) (*Listener, error) { @@ -1020,17 +1017,7 @@ func serveConn(block BlockCrypt, dataShards, parityShards int, conn net.PacketCo l.dataShards = dataShards l.parityShards = parityShards l.block = block - l.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards) l.chSocketReadError = make(chan struct{}) - - // calculate header size - if l.block != nil { - l.headerSize += cryptHeaderSize - } - if l.fecDecoder != nil { - l.headerSize += fecHeaderSizePlus2 - } - go l.monitor() return l, nil } @@ -1042,7 +1029,7 @@ func Dial(raddr string) (net.Conn, error) { return DialWithOptions(raddr, nil, 0 // // 'block' is the block encryption algorithm to encrypt packets. // -// 'dataShards', 'parityShards' specifiy how many parity packets will be generated following the data packets. +// 'dataShards', 'parityShards' specify how many parity packets will be generated following the data packets. // // Check https://github.com/klauspost/reedsolomon for details func DialWithOptions(raddr string, block BlockCrypt, dataShards, parityShards int) (*UDPSession, error) { diff --git a/vendor/github.com/xtaci/kcp-go/v5/snmp.go b/vendor/github.com/xtaci/kcp-go/v5/snmp.go index 607118e..f961810 100644 --- a/vendor/github.com/xtaci/kcp-go/v5/snmp.go +++ b/vendor/github.com/xtaci/kcp-go/v5/snmp.go @@ -25,7 +25,7 @@ type Snmp struct { RetransSegs uint64 // accmulated retransmited segments FastRetransSegs uint64 // accmulated fast retransmitted segments EarlyRetransSegs uint64 // accmulated early retransmitted segments - LostSegs uint64 // number of segs infered as lost + LostSegs uint64 // number of segs inferred as lost RepeatSegs uint64 // number of segs duplicated FECRecovered uint64 // correct packets recovered from FEC FECErrs uint64 // incorrect packets recovered from FEC diff --git a/vendor/modules.txt b/vendor/modules.txt index 038607c..eb23a42 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -38,7 +38,7 @@ github.com/tjfoc/gmsm/sm4 # github.com/urfave/cli v1.21.0 ## explicit github.com/urfave/cli -# github.com/xtaci/kcp-go/v5 v5.5.17 +# github.com/xtaci/kcp-go/v5 v5.6.1 ## explicit github.com/xtaci/kcp-go/v5 # github.com/xtaci/smux v1.5.14