diff --git a/fec.go b/fec.go index 1b41fcb..4b9b55d 100644 --- a/fec.go +++ b/fec.go @@ -285,9 +285,6 @@ type ( // RS encoder codec reedsolomon.Encoder - - // record min rto - minRTO int } ) @@ -298,7 +295,6 @@ func newFECEncoder(dataShards, parityShards, offset, minRTO int) *fecEncoder { enc := new(fecEncoder) enc.dataShards = dataShards enc.parityShards = parityShards - enc.minRTO = minRTO enc.shardSize = dataShards + parityShards enc.paws = 0xffffffff / uint32(enc.shardSize) * uint32(enc.shardSize) enc.headerOffset = offset @@ -322,7 +318,7 @@ func newFECEncoder(dataShards, parityShards, offset, minRTO int) *fecEncoder { // encodes the packet, outputs parity shards if we have collected quorum datashards // notice: the contents of 'ps' will be re-written in successive calling -func (enc *fecEncoder) encode(b []byte) (ps [][]byte) { +func (enc *fecEncoder) encode(b []byte, rto uint32) (ps [][]byte) { // The header format: // | FEC SEQID(4B) | FEC TYPE(2B) | SIZE (2B) | PAYLOAD(SIZE-2) | // |<-headerOffset |<-payloadOffset @@ -344,7 +340,7 @@ func (enc *fecEncoder) encode(b []byte) (ps [][]byte) { // Generation of Reed-Solomon Erasure Code if enc.shardCount == enc.dataShards { // generate the rs-code only if the data is continuous. - if enc.tsCache[enc.shardCount-1]-enc.tsCache[0] < int64(enc.minRTO) { + if enc.tsCache[enc.shardCount-1]-enc.tsCache[0] < int64(rto) { // fill '0' into the tail of each datashard for i := 0; i < enc.dataShards; i++ { shard := enc.shardCache[i] diff --git a/fec_test.go b/fec_test.go index 7e884e7..46a6a45 100644 --- a/fec_test.go +++ b/fec_test.go @@ -38,6 +38,6 @@ func BenchmarkFECEncode(b *testing.B) { encoder := newFECEncoder(dataSize, paritySize, 0, 200) for i := 0; i < b.N; i++ { data := make([]byte, payLoad) - encoder.encode(data) + encoder.encode(data, 200) } } diff --git a/readloop.go b/readloop.go deleted file mode 100644 index 697395a..0000000 --- a/readloop.go +++ /dev/null @@ -1,39 +0,0 @@ -package kcp - -import ( - "sync/atomic" - - "github.com/pkg/errors" -) - -func (s *UDPSession) defaultReadLoop() { - buf := make([]byte, mtuLimit) - var src string - for { - if n, addr, err := s.conn.ReadFrom(buf); err == nil { - // make sure the packet is from the same source - if src == "" { // set source address - src = addr.String() - } else if addr.String() != src { - atomic.AddUint64(&DefaultSnmp.InErrs, 1) - continue - } - s.packetInput(buf[:n]) - } else { - s.notifyReadError(errors.WithStack(err)) - return - } - } -} - -func (l *Listener) defaultMonitor() { - buf := make([]byte, mtuLimit) - for { - if n, from, err := l.conn.ReadFrom(buf); err == nil { - l.packetInput(buf[:n], from) - } else { - l.notifyReadError(errors.WithStack(err)) - return - } - } -} diff --git a/readloop_generic.go b/readloop_generic.go index 5dbe4f4..75a7125 100644 --- a/readloop_generic.go +++ b/readloop_generic.go @@ -1,11 +1,43 @@ +//go:build !linux // +build !linux package kcp -func (s *UDPSession) readLoop() { - s.defaultReadLoop() +import ( + "net" + "sync/atomic" + + "github.com/pkg/errors" +) + +func (s *UDPSession) readLoop(conn net.PacketConn) { + buf := make([]byte, mtuLimit) + var src string + for { + if n, addr, err := conn.ReadFrom(buf); err == nil { + // make sure the packet is from the same source + if src == "" { // set source address + src = addr.String() + } else if addr.String() != src { + atomic.AddUint64(&DefaultSnmp.InErrs, 1) + continue + } + s.packetInput(buf[:n]) + } else { + s.notifyReadError(errors.WithStack(err)) + return + } + } } -func (l *Listener) monitor() { - l.defaultMonitor() +func (l *Listener) monitor(conn net.PacketConn) { + buf := make([]byte, mtuLimit) + for { + if n, from, err := conn.ReadFrom(buf); err == nil { + l.packetInput(buf[:n], from) + } else { + l.notifyReadError(errors.WithStack(err)) + return + } + } } diff --git a/sess.go b/sess.go index 083f1e4..46c57ef 100644 --- a/sess.go +++ b/sess.go @@ -175,7 +175,7 @@ func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn sess.kcp.ReserveBytes(sess.headerSize) if sess.l == nil { // it's a client connection - go sess.readLoop() + go sess.readLoop(conn) atomic.AddUint64(&DefaultSnmp.ActiveOpens, 1) } else { atomic.AddUint64(&DefaultSnmp.PassiveOpens, 1) @@ -531,7 +531,7 @@ func (s *UDPSession) output(buf []byte) { // 1. FEC encoding if s.fecEncoder != nil { - ecc = s.fecEncoder.encode(buf) + ecc = s.fecEncoder.encode(buf, s.kcp.rx_rto) } // 2&3. crc32 & encryption @@ -1007,7 +1007,7 @@ func serveConn(block BlockCrypt, dataShards, parityShards int, conn net.PacketCo l.parityShards = parityShards l.block = block l.chSocketReadError = make(chan struct{}) - go l.monitor() + go l.monitor(conn) return l, nil } diff --git a/tx.go b/tx.go deleted file mode 100644 index 3397b82..0000000 --- a/tx.go +++ /dev/null @@ -1,24 +0,0 @@ -package kcp - -import ( - "sync/atomic" - - "github.com/pkg/errors" - "golang.org/x/net/ipv4" -) - -func (s *UDPSession) defaultTx(txqueue []ipv4.Message) { - nbytes := 0 - npkts := 0 - for k := range txqueue { - if n, err := s.conn.WriteTo(txqueue[k].Buffers[0], txqueue[k].Addr); err == nil { - nbytes += n - npkts++ - } else { - s.notifyWriteError(errors.WithStack(err)) - break - } - } - atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(npkts)) - atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes)) -} diff --git a/tx_generic.go b/tx_generic.go index 0b4f349..3e51e0a 100644 --- a/tx_generic.go +++ b/tx_generic.go @@ -1,11 +1,28 @@ +//go:build !linux // +build !linux package kcp import ( + "sync/atomic" + + "github.com/pkg/errors" "golang.org/x/net/ipv4" ) func (s *UDPSession) tx(txqueue []ipv4.Message) { - s.defaultTx(txqueue) + nbytes := 0 + npkts := 0 + for k := range txqueue { + if n, err := s.conn.WriteTo(txqueue[k].Buffers[0], txqueue[k].Addr); err == nil { + nbytes += n + npkts++ + } else { + s.notifyWriteError(errors.WithStack(err)) + break + } + } + atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(npkts)) + atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes)) + }