diff --git a/.travis.yml b/.travis.yml index 4003ab2..0e7c4ac 100644 --- a/.travis.yml +++ b/.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/autotune.go b/autotune.go new file mode 100644 index 0000000..e850ad3 --- /dev/null +++ b/autotune.go @@ -0,0 +1,66 @@ +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 { + // find left and right edges for a bit + lastPulse := tune.pulses[0] + idx := 1 + + // left edge + var leftEdge int + for ; idx < len(tune.pulses); idx++ { + if lastPulse.seq+1 == tune.pulses[idx].seq { + if lastPulse.bit != bit && tune.pulses[idx].bit == bit { + leftEdge = idx + break + } + } else { + return -1 + } + 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 { + if lastPulse.bit == bit && tune.pulses[idx].bit != bit { + rightEdge = idx + break + } + } else { + return -1 + } + lastPulse = tune.pulses[idx] + } + + return rightEdge - leftEdge +} diff --git a/autotune_test.go b/autotune_test.go new file mode 100644 index 0000000..3dc1ecc --- /dev/null +++ b/autotune_test.go @@ -0,0 +1,47 @@ +package kcp + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestAutoTune(t *testing.T) { + signals := []uint32{0, 0, 0, 0, 0, 0} + + tune := autoTune{} + for i := 0; i < len(signals); i++ { + if signals[i] == 0 { + tune.Sample(false, uint32(i)) + } else { + tune.Sample(true, uint32(i)) + } + } + + assert.Equal(t, -1, tune.FindPeriod(false)) + assert.Equal(t, -1, tune.FindPeriod(true)) + + signals = []uint32{1, 0, 1, 0, 0, 1} + tune = autoTune{} + for i := 0; i < len(signals); i++ { + if signals[i] == 0 { + tune.Sample(false, uint32(i)) + } else { + tune.Sample(true, uint32(i)) + } + } + assert.Equal(t, 1, tune.FindPeriod(false)) + assert.Equal(t, 1, tune.FindPeriod(true)) + + signals = []uint32{1, 0, 0, 0, 0, 1} + tune = autoTune{} + for i := 0; i < len(signals); i++ { + if signals[i] == 0 { + tune.Sample(false, uint32(i)) + } else { + tune.Sample(true, uint32(i)) + } + } + assert.Equal(t, -1, tune.FindPeriod(true)) + assert.Equal(t, 4, tune.FindPeriod(false)) +} diff --git a/fec.go b/fec.go index 97cd40b..1e33319 100644 --- a/fec.go +++ b/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,46 @@ 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 { + 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/fec_test.go b/fec_test.go index fffc69b..59b64ac 100644 --- a/fec_test.go +++ b/fec_test.go @@ -10,7 +10,7 @@ func BenchmarkFECDecode(b *testing.B) { const dataSize = 10 const paritySize = 3 const payLoad = 1500 - decoder := newFECDecoder(1024, dataSize, paritySize) + decoder := newFECDecoder(dataSize, paritySize) b.ReportAllocs() b.SetBytes(payLoad) for i := 0; i < b.N; i++ { diff --git a/go.mod b/go.mod index f80b2bf..ff51020 100644 --- a/go.mod +++ b/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/go.sum b/go.sum index ebb0cc1..db81c71 100644 --- a/go.sum +++ b/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/kcp.go b/kcp.go index 3d1984d..0c6c304 100644 --- a/kcp.go +++ b/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/kcp_test.go b/kcp_test.go index 435d6cc..49d55d5 100644 --- a/kcp_test.go +++ b/kcp_test.go @@ -104,7 +104,7 @@ func testlink(t *testing.T, client *lossyconn.LossyConn, server *lossyconn.Lossy start := time.Now() s.Write(buf) io.ReadFull(s, buf) - rtt += time.Now().Sub(start) + rtt += time.Since(start) } t.Log("client:", client) diff --git a/readloop.go b/readloop.go index bc48e84..697395a 100644 --- a/readloop.go +++ b/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/readloop_linux.go b/readloop_linux.go index 6226478..be194af 100644 --- a/readloop_linux.go +++ b/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/sess.go b/sess.go index e5ada2b..2dedd74 100644 --- a/sess.go +++ b/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/sess_test.go b/sess_test.go index 8c223d5..fbe3ad1 100644 --- a/sess_test.go +++ b/sess_test.go @@ -94,7 +94,7 @@ func listenEcho(port int) (net.Listener, error) { //block, _ := NewTEABlockCrypt(pass[:16]) //block, _ := NewAESBlockCrypt(pass) block, _ := NewSalsa20BlockCrypt(pass) - return ListenWithOptions(fmt.Sprintf("127.0.0.1:%v", port), block, 10, 3) + return ListenWithOptions(fmt.Sprintf("127.0.0.1:%v", port), block, 10, 0) } func listenTinyBufferEcho(port int) (net.Listener, error) { //block, _ := NewNoneBlockCrypt(pass) diff --git a/snmp.go b/snmp.go index 607118e..f961810 100644 --- a/snmp.go +++ b/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