upgrade to kcp-go@v5.6.1

2024-04-21 12:32:32 +00:00 · 2020-10-10 11:52:31 +08:00
parent b2be91cdb5
commit 114cd97025
14 changed files with 214 additions and 118 deletions
@@ -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
@@ -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
@@ -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=
@@ -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)
@@ -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
+}
@@ -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
@@ -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
@@ -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=
@@ -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)
@@ -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
@@ -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:
@@ -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) {
@@ -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
@@ -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