1. simplify readloop_xxx files & tx_xxx files by merging

2. fix the race in owned packet test

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]

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)
 	}
 }

readloop.go

@@ -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
-		}
-	}
-}

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
+		}
+	}
 }

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
 }
 

tx.go

@@ -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))
-}

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))
+
 }