Revert "1. simplify readloop_xxx files & tx_xxx files by merging"

This reverts commit 681bda64d1.

fec.go

@@ -285,6 +285,9 @@ type (
 
 		// RS encoder
 		codec reedsolomon.Encoder
+
+		// record min rto
+		minRTO int
 	}
 )
 
@@ -295,6 +298,7 @@ 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
@@ -318,7 +322,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, rto uint32) (ps [][]byte) {
+func (enc *fecEncoder) encode(b []byte) (ps [][]byte) {
 	// The header format:
 	// | FEC SEQID(4B) | FEC TYPE(2B) | SIZE (2B) | PAYLOAD(SIZE-2) |
 	// |<-headerOffset                |<-payloadOffset
@@ -340,7 +344,7 @@ func (enc *fecEncoder) encode(b []byte, rto uint32) (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(rto) {
+		if enc.tsCache[enc.shardCount-1]-enc.tsCache[0] < int64(enc.minRTO) {
 			// 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, 200)
+		encoder.encode(data)
 	}
 }

readloop.go

@@ -0,0 +1,39 @@
+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,43 +1,11 @@
-//go:build !linux
 // +build !linux
 
 package kcp
 
-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 (s *UDPSession) readLoop() {
+	s.defaultReadLoop()
 }
 
-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
-		}
-	}
+func (l *Listener) monitor() {
+	l.defaultMonitor()
 }

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(conn)
+		go sess.readLoop()
 		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, s.kcp.rx_rto)
+		ecc = s.fecEncoder.encode(buf)
 	}
 
 	// 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(conn)
+	go l.monitor()
 	return l, nil
 }
 

tx.go

@@ -0,0 +1,24 @@
+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,28 +1,11 @@
-//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) {
-	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))
-
+	s.defaultTx(txqueue)
 }