upd deps to smux to v1.5.16

* create udp connection only when needed

* misc

* some more minor changes

* update option description

* sync the doc

* lets keep original

* respect config.Quiet

* misc

* removed the useless muxes init

README.md

@@ -152,7 +152,7 @@ GLOBAL OPTIONS:
    --mode value                     profiles: fast3, fast2, fast, normal, manual (default: "fast")
    --conn value                     set num of UDP connections to server (default: 1)
    --autoexpire value               set auto expiration time(in seconds) for a single UDP connection, 0 to disable (default: 0)
-   --scavengettl value              set how long an expired connection can live(in sec), -1 to disable (default: 600)
+   --scavengettl value              set how long an expired connection can live (in seconds) (default: 600)
    --mtu value                      set maximum transmission unit for UDP packets (default: 1350)
    --sndwnd value                   set send window size(num of packets) (default: 128)
    --rcvwnd value                   set receive window size(num of packets) (default: 512)

build-release.sh

@@ -57,6 +57,12 @@ for os in ${OSES[@]}; do
 	$sum kcptun-${os}-386-$VERSION.tar.gz
 done
 
+#Apple M1 device
+env CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build -mod=vendor -ldflags "$LDFLAGS" -gcflags "$GCFLAGS" -o server_darwin_arm64 github.com/xtaci/kcptun/server
+env CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build -mod=vendor -ldflags "$LDFLAGS" -gcflags "$GCFLAGS" -o client_darwin_arm64 github.com/xtaci/kcptun/client
+tar -zcf kcptun-darwin-arm64-$VERSION.tar.gz client_darwin_arm64 server_darwin_arm64
+$sum kcptun-darwin-arm64-$VERSION.tar.gz
+
 # ARM
 ARMS=(5 6 7)
 for v in ${ARMS[@]}; do

client/main.go

@@ -73,6 +73,11 @@ func checkError(err error) {
 	}
 }
 
+type timedSession struct {
+	session    *smux.Session
+	expiryDate time.Time
+}
+
 func main() {
 	rand.Seed(int64(time.Now().Nanosecond()))
 	if VERSION == "SELFBUILD" {
@@ -124,7 +129,7 @@ func main() {
 		cli.IntFlag{
 			Name:  "scavengettl",
 			Value: 600,
-			Usage: "set how long an expired connection can live(in sec), -1 to disable",
+			Usage: "set how long an expired connection can live (in seconds)",
 		},
 		cli.IntFlag{
 			Name:  "mtu",
@@ -421,20 +426,16 @@ func main() {
 			}
 		}
 
-		numconn := uint16(config.Conn)
-		muxes := make([]struct {
-			session *smux.Session
-			ttl     time.Time
-		}, numconn)
-
-		for k := range muxes {
-			muxes[k].session = waitConn()
-			muxes[k].ttl = time.Now().Add(time.Duration(config.AutoExpire) * time.Second)
-		}
-
-		chScavenger := make(chan *smux.Session, 128)
-		go scavenger(chScavenger, config.ScavengeTTL)
+		// start snmp logger
 		go generic.SnmpLogger(config.SnmpLog, config.SnmpPeriod)
+
+		// start scavenger
+		chScavenger := make(chan timedSession, 128)
+		go scavenger(chScavenger, &config)
+
+		// start listener
+		numconn := uint16(config.Conn)
+		muxes := make([]timedSession, numconn)
 		rr := uint16(0)
 		for {
 			p1, err := listener.AcceptTCP()
@@ -444,10 +445,13 @@ func main() {
 			idx := rr % numconn
 
 			// do auto expiration && reconnection
-			if muxes[idx].session.IsClosed() || (config.AutoExpire > 0 && time.Now().After(muxes[idx].ttl)) {
-				chScavenger <- muxes[idx].session
+			if muxes[idx].session == nil || muxes[idx].session.IsClosed() ||
+				(config.AutoExpire > 0 && time.Now().After(muxes[idx].expiryDate)) {
 				muxes[idx].session = waitConn()
-				muxes[idx].ttl = time.Now().Add(time.Duration(config.AutoExpire) * time.Second)
+				muxes[idx].expiryDate = time.Now().Add(time.Duration(config.AutoExpire) * time.Second)
+				if config.AutoExpire > 0 { // only when autoexpire set
+					chScavenger <- muxes[idx]
+				}
 			}
 
 			go handleClient(muxes[idx].session, p1, config.Quiet)
@@ -457,30 +461,35 @@ func main() {
 	myApp.Run(os.Args)
 }
 
-type scavengeSession struct {
-	session *smux.Session
-	ts      time.Time
-}
+func scavenger(ch chan timedSession, config *Config) {
+	// When AutoExpire is set to 0 (default), sessionList will keep empty.
+	// Then this routine won't need to do anything; thus just terminate it.
+	if config.AutoExpire <= 0 {
+		return
+	}
 
-func scavenger(ch chan *smux.Session, ttl int) {
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
-	var sessionList []scavengeSession
+	var sessionList []timedSession
 	for {
 		select {
-		case sess := <-ch:
-			sessionList = append(sessionList, scavengeSession{sess, time.Now()})
-			log.Println("session marked as expired", sess.RemoteAddr())
+		case item := <-ch:
+			sessionList = append(sessionList, timedSession{
+				item.session,
+				item.expiryDate.Add(time.Duration(config.ScavengeTTL) * time.Second)})
 		case <-ticker.C:
-			var newList []scavengeSession
+			if len(sessionList) == 0 {
+				continue
+			}
+
+			var newList []timedSession
 			for k := range sessionList {
 				s := sessionList[k]
-				if s.session.NumStreams() == 0 || s.session.IsClosed() {
-					log.Println("session normally closed", s.session.RemoteAddr())
-					s.session.Close()
-				} else if ttl >= 0 && time.Since(s.ts) >= time.Duration(ttl)*time.Second {
-					log.Println("session reached scavenge ttl", s.session.RemoteAddr())
+				if s.session.IsClosed() {
+					log.Println("scavenger: session normally closed:", s.session.LocalAddr())
+				} else if time.Now().After(s.expiryDate) {
 					s.session.Close()
+					log.Println("scavenger: session closed due to ttl:", s.session.LocalAddr())
 				} else {
 					newList = append(newList, sessionList[k])
 				}

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/urfave/cli v1.21.0
 	github.com/xtaci/kcp-go/v5 v5.6.1
-	github.com/xtaci/smux v1.5.14
+	github.com/xtaci/smux v1.5.16
 	github.com/xtaci/tcpraw v1.2.25
 	golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de
 )

go.sum

@@ -40,6 +40,10 @@ github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae h1:J0GxkO96kL4WF+A
 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=
 github.com/xtaci/smux v1.5.14/go.mod h1:OMlQbT5vcgl2gb49mFkYo6SMf+zP3rcjcwQz7ZU7IGY=
+github.com/xtaci/smux v1.5.15 h1:6hMiXswcleXj5oNfcJc+DXS8Vj36XX2LaX98udog6Kc=
+github.com/xtaci/smux v1.5.15/go.mod h1:OMlQbT5vcgl2gb49mFkYo6SMf+zP3rcjcwQz7ZU7IGY=
+github.com/xtaci/smux v1.5.16 h1:FBPYOkW8ZTjLKUM4LI4xnnuuDC8CQ/dB04HD519WoEk=
+github.com/xtaci/smux v1.5.16/go.mod h1:OMlQbT5vcgl2gb49mFkYo6SMf+zP3rcjcwQz7ZU7IGY=
 github.com/xtaci/tcpraw v1.2.25 h1:VDlqo0op17JeXBM6e2G9ocCNLOJcw9mZbobMbJjo0vk=
 github.com/xtaci/tcpraw v1.2.25/go.mod h1:dKyZ2V75s0cZ7cbgJYdxPvms7af0joIeOyx1GgJQbLk=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=

vendor/github.com/xtaci/smux/.travis.yml

@@ -1,3 +1,6 @@
+arch:
+    - amd64
+    - ppc64le
 language: go
 go:
     - 1.9.x

vendor/github.com/xtaci/smux/README.md

@@ -5,7 +5,7 @@
 <img src="mux.jpg" alt="smux" height="120px" /> 
 
 [1]: https://godoc.org/github.com/xtaci/smux?status.svg
-[2]: https://pkg.go.dev/github.com/xtaci/smux
+[2]: https://godoc.org/github.com/xtaci/smux
 [3]: https://img.shields.io/badge/license-MIT-blue.svg
 [4]: LICENSE
 [5]: https://travis-ci.org/xtaci/smux.svg?branch=master
@@ -34,7 +34,7 @@ Smux ( **S**imple **MU**ltiple**X**ing) is a multiplexing library for Golang. It
 
 ## Documentation
 
-For complete documentation, see the associated [Godoc](https://pkg.go.dev/github.com/xtaci/smux).
+For complete documentation, see the associated [Godoc](https://godoc.org/github.com/xtaci/smux).
 
 ## Benchmark
 ```

vendor/github.com/xtaci/smux/mux.go

@@ -17,6 +17,9 @@ type Config struct {
 	// SMUX Protocol version, support 1,2
 	Version int
 
+	// Disabled keepalive
+	KeepAliveDisabled bool
+
 	// KeepAliveInterval is how often to send a NOP command to the remote
 	KeepAliveInterval time.Duration
 
@@ -54,11 +57,13 @@ func VerifyConfig(config *Config) error {
 	if !(config.Version == 1 || config.Version == 2) {
 		return errors.New("unsupported protocol version")
 	}
-	if config.KeepAliveInterval == 0 {
-		return errors.New("keep-alive interval must be positive")
-	}
-	if config.KeepAliveTimeout < config.KeepAliveInterval {
-		return fmt.Errorf("keep-alive timeout must be larger than keep-alive interval")
+	if !config.KeepAliveDisabled {
+		if config.KeepAliveInterval == 0 {
+			return errors.New("keep-alive interval must be positive")
+		}
+		if config.KeepAliveTimeout < config.KeepAliveInterval {
+			return fmt.Errorf("keep-alive timeout must be larger than keep-alive interval")
+		}
 	}
 	if config.MaxFrameSize <= 0 {
 		return errors.New("max frame size must be positive")

vendor/github.com/xtaci/smux/session.go

@@ -24,7 +24,7 @@ var (
 )
 
 type writeRequest struct {
-	prio   uint64
+	prio   uint32
 	frame  Frame
 	result chan writeResult
 }
@@ -104,7 +104,9 @@ func newSession(config *Config, conn io.ReadWriteCloser, client bool) *Session {
 	go s.shaperLoop()
 	go s.recvLoop()
 	go s.sendLoop()
-	go s.keepalive()
+	if !config.KeepAliveDisabled {
+		go s.keepalive()
+	}
 	return s
 }
 
@@ -494,7 +496,7 @@ func (s *Session) writeFrame(f Frame) (n int, err error) {
 }
 
 // internal writeFrame version to support deadline used in keepalive
-func (s *Session) writeFrameInternal(f Frame, deadline <-chan time.Time, prio uint64) (int, error) {
+func (s *Session) writeFrameInternal(f Frame, deadline <-chan time.Time, prio uint32) (int, error) {
 	req := writeRequest{
 		prio:   prio,
 		frame:  f,

vendor/github.com/xtaci/smux/shaper.go

@@ -1,9 +1,13 @@
 package smux
 
+func _itimediff(later, earlier uint32) int32 {
+	return (int32)(later - earlier)
+}
+
 type shaperHeap []writeRequest
 
 func (h shaperHeap) Len() int            { return len(h) }
-func (h shaperHeap) Less(i, j int) bool  { return h[i].prio < h[j].prio }
+func (h shaperHeap) Less(i, j int) bool  { return _itimediff(h[j].prio, h[i].prio) > 0 }
 func (h shaperHeap) Swap(i, j int)       { h[i], h[j] = h[j], h[i] }
 func (h *shaperHeap) Push(x interface{}) { *h = append(*h, x.(writeRequest)) }
 

vendor/github.com/xtaci/smux/stream.go

@@ -272,6 +272,12 @@ func (s *Stream) waitRead() error {
 	case <-s.chReadEvent:
 		return nil
 	case <-s.chFinEvent:
+		// BUG(xtaci): Fix for https://github.com/xtaci/smux/issues/82
+		s.bufferLock.Lock()
+		defer s.bufferLock.Unlock()
+		if len(s.buffers) > 0 {
+			return nil
+		}
 		return io.EOF
 	case <-s.sess.chSocketReadError:
 		return s.sess.socketReadError.Load().(error)
@@ -319,7 +325,7 @@ func (s *Stream) Write(b []byte) (n int, err error) {
 		}
 		frame.data = bts[:sz]
 		bts = bts[sz:]
-		n, err := s.sess.writeFrameInternal(frame, deadline, uint64(s.numWritten))
+		n, err := s.sess.writeFrameInternal(frame, deadline, s.numWritten)
 		s.numWritten++
 		sent += n
 		if err != nil {
@@ -387,7 +393,7 @@ func (s *Stream) writeV2(b []byte) (n int, err error) {
 				}
 				frame.data = bts[:sz]
 				bts = bts[sz:]
-				n, err := s.sess.writeFrameInternal(frame, deadline, uint64(atomic.LoadUint32(&s.numWritten)))
+				n, err := s.sess.writeFrameInternal(frame, deadline, atomic.LoadUint32(&s.numWritten))
 				atomic.AddUint32(&s.numWritten, uint32(sz))
 				sent += n
 				if err != nil {

vendor/modules.txt

@@ -41,7 +41,7 @@ github.com/urfave/cli
 # github.com/xtaci/kcp-go/v5 v5.6.1
 ## explicit
 github.com/xtaci/kcp-go/v5
-# github.com/xtaci/smux v1.5.14
+# github.com/xtaci/smux v1.5.16
 ## explicit
 github.com/xtaci/smux
 # github.com/xtaci/tcpraw v1.2.25