add -log option to redirect log , see issue #199

This reverts commit 2f479b788c.

client/config.go

@@ -5,6 +5,7 @@ import (
 	"os"
 )
 
+// Config for client
 type Config struct {
 	LocalAddr    string `json:"localaddr"`
 	RemoteAddr   string `json:"remoteaddr"`
@@ -27,18 +28,15 @@ type Config struct {
 	NoCongestion int    `json:"nc"`
 	SockBuf      int    `json:"sockbuf"`
 	KeepAlive    int    `json:"keepalive"`
+	Log          string `json:"log"`
 }
 
-func parseJsonConfig(config *Config, path string) error {
+func parseJSONConfig(config *Config, path string) error {
 	file, err := os.Open(path) // For read access.
 	if err != nil {
 		return err
 	}
 	defer file.Close()
 
-	if err = json.NewDecoder(file).Decode(config); err != nil {
-		return err
-	}
-
-	return err
+	return json.NewDecoder(file).Decode(config)
 }

client/main.go

@@ -7,15 +7,15 @@ import (
 	"math/rand"
 	"net"
 	"os"
-	"runtime"
 	"time"
 
 	"golang.org/x/crypto/pbkdf2"
 
 	"github.com/golang/snappy"
+	"github.com/pkg/errors"
 	"github.com/urfave/cli"
 	kcp "github.com/xtaci/kcp-go"
-	"github.com/xtaci/yamux"
+	"github.com/xtaci/smux"
 )
 
 var (
@@ -61,16 +61,10 @@ func handleClient(p1, p2 io.ReadWriteCloser) {
 
 	// start tunnel
 	p1die := make(chan struct{})
-	go func() {
-		io.Copy(p1, p2)
-		close(p1die)
-	}()
+	go func() { io.Copy(p1, p2); close(p1die) }()
 
 	p2die := make(chan struct{})
-	go func() {
-		io.Copy(p2, p1)
-		close(p2die)
-	}()
+	go func() { io.Copy(p2, p1); close(p2die) }()
 
 	// wait for tunnel termination
 	select {
@@ -94,7 +88,7 @@ func main() {
 	}
 	myApp := cli.NewApp()
 	myApp.Name = "kcptun"
-	myApp.Usage = "kcptun client"
+	myApp.Usage = "client(with SMUX)"
 	myApp.Version = VERSION
 	myApp.Flags = []cli.Flag{
 		cli.StringFlag{
@@ -202,6 +196,11 @@ func main() {
 			Value:  10, // nat keepalive interval in seconds
 			Hidden: true,
 		},
+		cli.StringFlag{
+			Name:  "log",
+			Value: "",
+			Usage: "specify a log file to output, default goes to stderr",
+		},
 		cli.StringFlag{
 			Name:  "c",
 			Value: "", // when the value is not empty, the config path must exists
@@ -231,12 +230,21 @@ func main() {
 		config.NoCongestion = c.Int("nc")
 		config.SockBuf = c.Int("sockbuf")
 		config.KeepAlive = c.Int("keepalive")
+		config.Log = c.String("log")
 
 		if c.String("c") != "" {
-			err := parseJsonConfig(&config, c.String("c"))
+			err := parseJSONConfig(&config, c.String("c"))
 			checkError(err)
 		}
 
+		// log redirect
+		if config.Log != "" {
+			f, err := os.OpenFile(config.Log, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
+			checkError(err)
+			defer f.Close()
+			log.SetOutput(f)
+		}
+
 		switch config.Mode {
 		case "normal":
 			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 0, 30, 2, 1
@@ -299,17 +307,14 @@ func main() {
 		log.Println("conn:", config.Conn)
 		log.Println("autoexpire:", config.AutoExpire)
 
-		yconfig := &yamux.Config{
-			AcceptBacklog:          256,
-			EnableKeepAlive:        true,
-			KeepAliveInterval:      30 * time.Second,
-			ConnectionWriteTimeout: 10 * time.Second,
-			MaxStreamWindowSize:    uint32(config.SockBuf),
-			LogOutput:              os.Stderr,
-		}
-		createConn := func() *yamux.Session {
+		smuxConfig := smux.DefaultConfig()
+		smuxConfig.MaxReceiveBuffer = config.SockBuf
+
+		createConn := func() (*smux.Session, error) {
 			kcpconn, err := kcp.DialWithOptions(config.RemoteAddr, block, config.DataShard, config.ParityShard)
-			checkError(err)
+			if err != nil {
+				return nil, errors.Wrap(err, "createConn()")
+			}
 			kcpconn.SetStreamMode(true)
 			kcpconn.SetNoDelay(config.NoDelay, config.Interval, config.Resend, config.NoCongestion)
 			kcpconn.SetWindowSize(config.SndWnd, config.RcvWnd)
@@ -328,30 +333,44 @@ func main() {
 			}
 
 			// stream multiplex
-			var session *yamux.Session
+			var session *smux.Session
 			if config.NoComp {
-				session, err = yamux.Client(kcpconn, yconfig)
+				session, err = smux.Client(kcpconn, smuxConfig)
 			} else {
-				session, err = yamux.Client(newCompStream(kcpconn), yconfig)
+				session, err = smux.Client(newCompStream(kcpconn), smuxConfig)
+			}
+			if err != nil {
+				return nil, errors.Wrap(err, "createConn()")
+			}
+			return session, nil
+		}
+
+		// wait until a connection is ready
+		waitConn := func() *smux.Session {
+			for {
+				if session, err := createConn(); err == nil {
+					return session
+				} else {
+					time.Sleep(time.Second)
+				}
 			}
-			checkError(err)
-			runtime.SetFinalizer(session, func(s *yamux.Session) {
-				s.Close()
-			})
-			return session
 		}
 
 		numconn := uint16(config.Conn)
 		muxes := make([]struct {
-			session *yamux.Session
+			session *smux.Session
 			ttl     time.Time
 		}, numconn)
 
 		for k := range muxes {
-			muxes[k].session = createConn()
+			sess, err := createConn()
+			checkError(err)
+			muxes[k].session = sess
 			muxes[k].ttl = time.Now().Add(time.Duration(config.AutoExpire) * time.Second)
 		}
 
+		chScavenger := make(chan *smux.Session, 128)
+		go scavenger(chScavenger)
 		rr := uint16(0)
 		for {
 			p1, err := listener.AcceptTCP()
@@ -367,15 +386,16 @@ func main() {
 		OPEN_P2:
 			// do auto expiration
 			if config.AutoExpire > 0 && time.Now().After(muxes[idx].ttl) {
-				log.Println("autoexpired")
-				muxes[idx].session = createConn()
+				chScavenger <- muxes[idx].session
+				muxes[idx].session = waitConn()
 				muxes[idx].ttl = time.Now().Add(time.Duration(config.AutoExpire) * time.Second)
 			}
 
 			// do session open
-			p2, err := muxes[idx].session.Open()
-			if err != nil { // yamux failure
-				muxes[idx].session = createConn()
+			p2, err := muxes[idx].session.OpenStream()
+			if err != nil { // mux failure
+				chScavenger <- muxes[idx].session
+				muxes[idx].session = waitConn()
 				muxes[idx].ttl = time.Now().Add(time.Duration(config.AutoExpire) * time.Second)
 				goto OPEN_P2
 			}
@@ -385,3 +405,36 @@ func main() {
 	}
 	myApp.Run(os.Args)
 }
+
+type scavengeSession struct {
+	session *smux.Session
+	ttl     time.Time
+}
+
+const (
+	maxScavengeTTL = 10 * time.Minute
+)
+
+func scavenger(ch chan *smux.Session) {
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+	var sessionList []scavengeSession
+	for {
+		select {
+		case sess := <-ch:
+			sessionList = append(sessionList, scavengeSession{sess, time.Now()})
+		case <-ticker.C:
+			var newList []scavengeSession
+			for k := range sessionList {
+				s := sessionList[k]
+				if s.session.NumStreams() == 0 || s.session.IsClosed() || time.Since(s.ttl) > maxScavengeTTL {
+					log.Println("session scavenged")
+					s.session.Close()
+				} else {
+					newList = append(newList, sessionList[k])
+				}
+			}
+			sessionList = newList
+		}
+	}
+}

server/config.go

@@ -5,6 +5,7 @@ import (
 	"os"
 )
 
+// Config for server
 type Config struct {
 	Listen       string `json:"listen"`
 	Target       string `json:"target"`
@@ -25,18 +26,15 @@ type Config struct {
 	NoCongestion int    `json:"nc"`
 	SockBuf      int    `json:"sockbuf"`
 	KeepAlive    int    `json:"keepalive"`
+	Log          string `json:"log"`
 }
 
-func parseJsonConfig(config *Config, path string) error {
+func parseJSONConfig(config *Config, path string) error {
 	file, err := os.Open(path) // For read access.
 	if err != nil {
 		return err
 	}
 	defer file.Close()
 
-	if err = json.NewDecoder(file).Decode(config); err != nil {
-		return err
-	}
-
-	return err
+	return json.NewDecoder(file).Decode(config)
 }

server/main.go

@@ -14,7 +14,7 @@ import (
 	"github.com/golang/snappy"
 	"github.com/urfave/cli"
 	kcp "github.com/xtaci/kcp-go"
-	"github.com/xtaci/yamux"
+	"github.com/xtaci/smux"
 )
 
 var (
@@ -53,35 +53,34 @@ func newCompStream(conn net.Conn) *compStream {
 }
 
 // handle multiplex-ed connection
-func handleMux(conn io.ReadWriteCloser, target string, config *yamux.Config) {
+func handleMux(conn io.ReadWriteCloser, config *Config) {
 	// stream multiplex
-	mux, err := yamux.Server(conn, config)
+	smuxConfig := smux.DefaultConfig()
+	smuxConfig.MaxReceiveBuffer = config.SockBuf
+	mux, err := smux.Server(conn, smuxConfig)
 	if err != nil {
 		log.Println(err)
 		return
 	}
 	defer mux.Close()
 	for {
-		p1, err := mux.Accept()
+		p1, err := mux.AcceptStream()
 		if err != nil {
 			log.Println(err)
 			return
 		}
-		sockbuf := int(config.MaxStreamWindowSize)
-		p2, err := net.DialTimeout("tcp", target, 5*time.Second)
+		p2, err := net.DialTimeout("tcp", config.Target, 5*time.Second)
 		if err != nil {
 			p1.Close()
 			log.Println(err)
 			continue
 		}
-
-		if err := p2.(*net.TCPConn).SetReadBuffer(sockbuf); err != nil {
+		if err := p2.(*net.TCPConn).SetReadBuffer(config.SockBuf); err != nil {
 			log.Println("TCP SetReadBuffer:", err)
 		}
-		if err := p2.(*net.TCPConn).SetWriteBuffer(sockbuf); err != nil {
+		if err := p2.(*net.TCPConn).SetWriteBuffer(config.SockBuf); err != nil {
 			log.Println("TCP SetWriteBuffer:", err)
 		}
-
 		go handleClient(p1, p2)
 	}
 }
@@ -94,16 +93,10 @@ func handleClient(p1, p2 io.ReadWriteCloser) {
 
 	// start tunnel
 	p1die := make(chan struct{})
-	go func() {
-		io.Copy(p1, p2)
-		close(p1die)
-	}()
+	go func() { io.Copy(p1, p2); close(p1die) }()
 
 	p2die := make(chan struct{})
-	go func() {
-		io.Copy(p2, p1)
-		close(p2die)
-	}()
+	go func() { io.Copy(p2, p1); close(p2die) }()
 
 	// wait for tunnel termination
 	select {
@@ -127,7 +120,7 @@ func main() {
 	}
 	myApp := cli.NewApp()
 	myApp.Name = "kcptun"
-	myApp.Usage = "kcptun server"
+	myApp.Usage = "server(with SMUX)"
 	myApp.Version = VERSION
 	myApp.Flags = []cli.Flag{
 		cli.StringFlag{
@@ -225,6 +218,11 @@ func main() {
 			Value:  10, // nat keepalive interval in seconds
 			Hidden: true,
 		},
+		cli.StringFlag{
+			Name:  "log",
+			Value: "",
+			Usage: "specify a log file to output, default goes to stderr",
+		},
 		cli.StringFlag{
 			Name:  "c",
 			Value: "", // when the value is not empty, the config path must exists
@@ -252,13 +250,22 @@ func main() {
 		config.NoCongestion = c.Int("nc")
 		config.SockBuf = c.Int("sockbuf")
 		config.KeepAlive = c.Int("keepalive")
+		config.Log = c.String("log")
 
 		if c.String("c") != "" {
 			//Now only support json config file
-			err := parseJsonConfig(&config, c.String("c"))
+			err := parseJSONConfig(&config, c.String("c"))
 			checkError(err)
 		}
 
+		// log redirect
+		if config.Log != "" {
+			f, err := os.OpenFile(config.Log, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
+			checkError(err)
+			defer f.Close()
+			log.SetOutput(f)
+		}
+
 		switch config.Mode {
 		case "normal":
 			config.NoDelay, config.Interval, config.Resend, config.NoCongestion = 0, 30, 2, 1
@@ -325,14 +332,6 @@ func main() {
 		if err := lis.SetWriteBuffer(config.SockBuf); err != nil {
 			log.Println("SetWriteBuffer:", err)
 		}
-		yconfig := &yamux.Config{
-			AcceptBacklog:          256,
-			EnableKeepAlive:        true,
-			KeepAliveInterval:      30 * time.Second,
-			ConnectionWriteTimeout: 10 * time.Second,
-			MaxStreamWindowSize:    uint32(config.SockBuf),
-			LogOutput:              os.Stderr,
-		}
 		for {
 			if conn, err := lis.AcceptKCP(); err == nil {
 				log.Println("remote address:", conn.RemoteAddr())
@@ -344,9 +343,9 @@ func main() {
 				conn.SetKeepAlive(config.KeepAlive)
 
 				if config.NoComp {
-					go handleMux(conn, config.Target, yconfig)
+					go handleMux(conn, &config)
 				} else {
-					go handleMux(newCompStream(conn), config.Target, yconfig)
+					go handleMux(newCompStream(conn), &config)
 				}
 			} else {
 				log.Printf("%+v", err)