refactor: refactor code (#373)

cmd/kubevpn/cmds/clone.go

@@ -160,7 +160,7 @@ func CmdClone(f cmdutil.Factory) *cobra.Command {
 	cmd.Flags().BoolVar(&config.Debug, "debug", false, "Enable debug mode or not, true or false")
 	cmd.Flags().StringVar(&config.Image, "image", config.Image, "Use this image to startup container")
 	cmd.Flags().BoolVar(&transferImage, "transfer-image", false, "transfer image to remote registry, it will transfer image "+config.OriginImage+" to flags `--image` special image, default: "+config.Image)
-	cmd.Flags().StringVar((*string)(&options.Engine), "netstack", string(config.EngineSystem), fmt.Sprintf(`network stack ("%s"|"%s") %s: use gvisor (both performance and stable), %s: use raw mode (best stable)`, config.EngineGvisor, config.EngineSystem, config.EngineGvisor, config.EngineSystem))
+	cmd.Flags().StringVar((*string)(&options.Engine), "netstack", string(config.EngineSystem), fmt.Sprintf(`network stack ("%s"|"%s") %s: use gvisor (good compatibility), %s: use raw mode (best performance, relays on iptables SNAT)`, config.EngineGvisor, config.EngineSystem, config.EngineGvisor, config.EngineSystem))
 
 	cmd.Flags().StringVar(&options.TargetImage, "target-image", "", "Clone container use this image to startup container, if not special, use origin image")
 	cmd.Flags().StringVar(&options.TargetContainer, "target-container", "", "Clone container use special image to startup this container, if not special, use origin image")

cmd/kubevpn/cmds/connect.go

@@ -162,7 +162,7 @@ func CmdConnect(f cmdutil.Factory) *cobra.Command {
 	cmd.Flags().BoolVar(&config.Debug, "debug", false, "enable debug mode or not, true or false")
 	cmd.Flags().StringVar(&config.Image, "image", config.Image, "use this image to startup container")
 	cmd.Flags().BoolVar(&transferImage, "transfer-image", false, "transfer image to remote registry, it will transfer image "+config.OriginImage+" to flags `--image` special image, default: "+config.Image)
-	cmd.Flags().StringVar((*string)(&connect.Engine), "netstack", string(config.EngineSystem), fmt.Sprintf(`network stack ("%s"|"%s") %s: use gvisor (both performance and stable), %s: use raw mode (best stable)`, config.EngineGvisor, config.EngineSystem, config.EngineGvisor, config.EngineSystem))
+	cmd.Flags().StringVar((*string)(&connect.Engine), "netstack", string(config.EngineSystem), fmt.Sprintf(`network stack ("%s"|"%s") %s: use gvisor (good compatibility), %s: use raw mode (best performance, relays on iptables SNAT)`, config.EngineGvisor, config.EngineSystem, config.EngineGvisor, config.EngineSystem))
 	cmd.Flags().BoolVar(&foreground, "foreground", false, "Hang up")
 	cmd.Flags().BoolVar(&lite, "lite", false, "connect to multiple cluster in lite mode. mode \"lite\": design for only connecting to multiple cluster network. mode \"full\": not only connect to cluster network, it also supports proxy workloads inbound traffic to local PC.")
 

cmd/kubevpn/cmds/controlplane.go

@@ -1,9 +1,6 @@
 package cmds
 
 import (
-	"context"
-	"time"
-
 	"github.com/docker/docker/libnetwork/resolvconf"
 	miekgdns "github.com/miekg/dns"
 	log "github.com/sirupsen/logrus"
@@ -33,16 +30,13 @@ func CmdControlPlane(_ cmdutil.Factory) *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			util.InitLoggerForServer(config.Debug)
 			go util.StartupPProfForServer(0)
-			go func(ctx context.Context) {
+			go func() {
 				conf, err := miekgdns.ClientConfigFromFile(resolvconf.Path())
 				if err != nil {
-					return
+					log.Fatal(err)
 				}
-				for ctx.Err() == nil {
+				log.Fatal(dns.ListenAndServe("udp", ":53", conf))
-					dns.ListenAndServe("udp", ":53", conf)
+			}()
-					time.Sleep(time.Second * 5)
-				}
-			}(cmd.Context())
 			err := controlplane.Main(cmd.Context(), watchDirectoryFilename, port, log.StandardLogger())
 			return err
 		},

cmd/kubevpn/cmds/dev.go

@@ -140,7 +140,7 @@ func CmdDev(f cmdutil.Factory) *cobra.Command {
 	cmdutil.CheckErr(cmd.RegisterFlagCompletionFunc("container", completion.ContainerCompletionFunc(f)))
 	cmd.Flags().StringVar((*string)(&options.ConnectMode), "connect-mode", string(dev.ConnectModeHost), "Connect to kubernetes network in container or in host, eg: ["+string(dev.ConnectModeContainer)+"|"+string(dev.ConnectModeHost)+"]")
 	cmd.Flags().BoolVar(&transferImage, "transfer-image", false, "transfer image to remote registry, it will transfer image "+config.OriginImage+" to flags `--image` special image, default: "+config.Image)
-	cmd.Flags().StringVar((*string)(&options.Engine), "netstack", string(config.EngineSystem), fmt.Sprintf(`network stack ("%s"|"%s") %s: use gvisor (both performance and stable), %s: use raw mode (best stable)`, config.EngineGvisor, config.EngineSystem, config.EngineGvisor, config.EngineSystem))
+	cmd.Flags().StringVar((*string)(&options.Engine), "netstack", string(config.EngineSystem), fmt.Sprintf(`network stack ("%s"|"%s") %s: use gvisor (good compatibility), %s: use raw mode (best performance, relays on iptables SNAT)`, config.EngineGvisor, config.EngineSystem, config.EngineGvisor, config.EngineSystem))
 
 	// diy docker options
 	cmd.Flags().StringVar(&options.DevImage, "dev-image", "", "Use to startup docker container, Default is pod image")

cmd/kubevpn/cmds/proxy.go

@@ -182,7 +182,7 @@ func CmdProxy(f cmdutil.Factory) *cobra.Command {
 	cmd.Flags().BoolVar(&config.Debug, "debug", false, "Enable debug mode or not, true or false")
 	cmd.Flags().StringVar(&config.Image, "image", config.Image, "Use this image to startup container")
 	cmd.Flags().BoolVar(&transferImage, "transfer-image", false, "transfer image to remote registry, it will transfer image "+config.OriginImage+" to flags `--image` special image, default: "+config.Image)
-	cmd.Flags().StringVar((*string)(&connect.Engine), "netstack", string(config.EngineSystem), fmt.Sprintf(`network stack ("%s"|"%s") %s: use gvisor (both performance and stable), %s: use raw mode (best stable)`, config.EngineGvisor, config.EngineSystem, config.EngineGvisor, config.EngineSystem))
+	cmd.Flags().StringVar((*string)(&connect.Engine), "netstack", string(config.EngineSystem), fmt.Sprintf(`network stack ("%s"|"%s") %s: use gvisor (good compatibility), %s: use raw mode (best performance, relays on iptables SNAT)`, config.EngineGvisor, config.EngineSystem, config.EngineGvisor, config.EngineSystem))
 	cmd.Flags().BoolVar(&foreground, "foreground", false, "foreground hang up")
 
 	handler.AddExtraRoute(cmd.Flags(), extraRoute)

cmd/kubevpn/cmds/serve.go

@@ -34,7 +34,7 @@ func CmdServe(_ cmdutil.Factory) *cobra.Command {
 		PreRun: func(*cobra.Command, []string) {
 			util.InitLoggerForServer(config.Debug)
 			runtime.GOMAXPROCS(0)
-			go util.StartupPProfForServer(6060)
+			go util.StartupPProfForServer(config.PProfPort)
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
 			rand.Seed(time.Now().UnixNano())

pkg/config/config.go

@@ -84,8 +84,6 @@ const (
 
 	EnvSSHJump = "SSH_JUMP_BY_KUBEVPN"
 
-	// transport mode
-	ConfigKubeVPNTransportEngine = "transport-engine"
 	// hosts entry key word
 	HostsKeyWord = "# Add by KubeVPN"
 )
@@ -130,9 +128,9 @@ func init() {
 var Debug bool
 
 var (
-	SmallBufferSize  = (1 << 13) - 1 // 8KB small buffer
+	SmallBufferSize  = 2 * 1024  // 2KB small buffer
-	MediumBufferSize = (1 << 15) - 1 // 32KB medium buffer
+	MediumBufferSize = 8 * 1024  // 8KB medium buffer
-	LargeBufferSize  = (1 << 16) - 1 // 64KB large buffer
+	LargeBufferSize  = 32 * 1024 // 32KB large buffer
 )
 
 var (
@@ -152,19 +150,23 @@ var (
 )
 
 var (
-	LPool = &sync.Pool{
+	SPool = &sync.Pool{
 		New: func() interface{} {
+			return make([]byte, SmallBufferSize)
+		},
+	}
+	MPool = sync.Pool{
+		New: func() any {
+			return make([]byte, MediumBufferSize)
+		},
+	}
+	LPool = sync.Pool{
+		New: func() any {
 			return make([]byte, LargeBufferSize)
 		},
 	}
 )
 
-var SPool = sync.Pool{
-	New: func() any {
-		return make([]byte, 2)
-	},
-}
-
 type Engine string
 
 const (

pkg/core/gvisortcpforwarder.go

@@ -4,12 +4,12 @@ import (
 	"bytes"
 	"context"
 	"encoding/binary"
-	"errors"
 	"fmt"
 	"io"
 	"net"
 	"time"
 
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
@@ -51,16 +51,16 @@ func TCPForwarder(s *stack.Stack, ctx context.Context) func(stack.TransportEndpo
 		defer remote.Close()
 		errChan := make(chan error, 2)
 		go func() {
-			i := config.LPool.Get().([]byte)[:]
+			buf := config.LPool.Get().([]byte)[:]
-			defer config.LPool.Put(i[:])
+			defer config.LPool.Put(buf[:])
-			written, err2 := io.CopyBuffer(remote, conn, i)
+			written, err2 := io.CopyBuffer(remote, conn, buf)
 			log.Debugf("[TUN-TCP] Write length %d data to remote", written)
 			errChan <- err2
 		}()
 		go func() {
-			i := config.LPool.Get().([]byte)[:]
+			buf := config.LPool.Get().([]byte)[:]
-			defer config.LPool.Put(i[:])
+			defer config.LPool.Put(buf[:])
-			written, err2 := io.CopyBuffer(conn, remote, i)
+			written, err2 := io.CopyBuffer(conn, remote, buf)
 			log.Debugf("[TUN-TCP] Read length %d data from remote", written)
 			errChan <- err2
 		}()
@@ -73,8 +73,8 @@ func TCPForwarder(s *stack.Stack, ctx context.Context) func(stack.TransportEndpo
 
 func WriteProxyInfo(conn net.Conn, id stack.TransportEndpointID) error {
 	var b bytes.Buffer
-	i := config.SPool.Get().([]byte)[:]
+	i := config.MPool.Get().([]byte)[:]
-	defer config.SPool.Put(i[:])
+	defer config.MPool.Put(i[:])
 	binary.BigEndian.PutUint16(i, id.LocalPort)
 	b.Write(i)
 	binary.BigEndian.PutUint16(i, id.RemotePort)

pkg/core/gvisortunendpoint.go

@@ -2,10 +2,7 @@ package core
 
 import (
 	"context"
-	"errors"
-	"io"
 	"net"
-	"os"
 
 	"github.com/google/gopacket/layers"
 	log "github.com/sirupsen/logrus"
@@ -35,14 +32,6 @@ func (h *gvisorTCPHandler) readFromEndpointWriteToTCPConn(ctx context.Context, c
 			buf := pktBuffer.ToView().AsSlice()
 			_, err := tcpConn.Write(buf)
 			if err != nil {
-				if errors.Is(err, os.ErrClosed) || errors.Is(err, io.EOF) {
-					return
-				}
-				// if context is done
-				if ctx.Err() != nil {
-					log.Errorf("[TUN] Failed to write to tun: %v, context is done: %v", err, ctx.Err())
-					return
-				}
 				log.Errorf("[TUN] Failed to write data to tun device: %v", err)
 			}
 		}
@@ -53,24 +42,22 @@ func (h *gvisorTCPHandler) readFromEndpointWriteToTCPConn(ctx context.Context, c
 func (h *gvisorTCPHandler) readFromTCPConnWriteToEndpoint(ctx context.Context, conn net.Conn, endpoint *channel.Endpoint) {
 	tcpConn, _ := newGvisorFakeUDPTunnelConnOverTCP(ctx, conn)
 	for {
-		bytes := config.LPool.Get().([]byte)[:]
+		select {
-		read, err := tcpConn.Read(bytes[:])
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		buf := config.SPool.Get().([]byte)[:]
+		read, err := tcpConn.Read(buf[:])
 		if err != nil {
-			if errors.Is(err, os.ErrClosed) || errors.Is(err, io.EOF) {
-				return
-			}
-			// if context is done
-			if ctx.Err() != nil {
-				log.Errorf("[TUN] Failed to read from tun: %v, context is done", err)
-				return
-			}
 			log.Errorf("[TUN] Failed to read from tcp conn: %v", err)
-			config.LPool.Put(bytes[:])
+			config.SPool.Put(buf[:])
 			continue
 		}
 		if read == 0 {
 			log.Warnf("[TUN] Read from tcp conn length is %d", read)
-			config.LPool.Put(bytes[:])
+			config.SPool.Put(buf[:])
 			continue
 		}
 		// Try to determine network protocol number, default zero.
@@ -79,23 +66,23 @@ func (h *gvisorTCPHandler) readFromTCPConnWriteToEndpoint(ctx context.Context, c
 		var src, dst net.IP
 		// TUN interface with IFF_NO_PI enabled, thus
 		// we need to determine protocol from version field
-		if util.IsIPv4(bytes) {
+		if util.IsIPv4(buf) {
 			protocol = header.IPv4ProtocolNumber
-			ipHeader, err := ipv4.ParseHeader(bytes[:read])
+			ipHeader, err := ipv4.ParseHeader(buf[:read])
 			if err != nil {
 				log.Errorf("Failed to parse IPv4 header: %v", err)
-				config.LPool.Put(bytes[:])
+				config.SPool.Put(buf[:])
 				continue
 			}
 			ipProtocol = ipHeader.Protocol
 			src = ipHeader.Src
 			dst = ipHeader.Dst
-		} else if util.IsIPv6(bytes) {
+		} else if util.IsIPv6(buf) {
 			protocol = header.IPv6ProtocolNumber
-			ipHeader, err := ipv6.ParseHeader(bytes[:read])
+			ipHeader, err := ipv6.ParseHeader(buf[:read])
 			if err != nil {
 				log.Errorf("Failed to parse IPv6 header: %s", err.Error())
-				config.LPool.Put(bytes[:])
+				config.SPool.Put(buf[:])
 				continue
 			}
 			ipProtocol = ipHeader.NextHeader
@@ -103,7 +90,7 @@ func (h *gvisorTCPHandler) readFromTCPConnWriteToEndpoint(ctx context.Context, c
 			dst = ipHeader.Dst
 		} else {
 			log.Debugf("[TUN-GVISOR] Unknown packet")
-			config.LPool.Put(bytes[:])
+			config.SPool.Put(buf[:])
 			continue
 		}
 
@@ -113,18 +100,18 @@ func (h *gvisorTCPHandler) readFromTCPConnWriteToEndpoint(ctx context.Context, c
 			log.Tracef("[TUN-RAW] Forward to TUN device, SRC: %s, DST: %s, Length: %d", src.String(), dst.String(), read)
 			util.SafeWrite(h.packetChan, &datagramPacket{
 				DataLength: uint16(read),
-				Data:       bytes[:],
+				Data:       buf[:],
 			})
 			continue
 		}
 
 		pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
 			ReserveHeaderBytes: 0,
-			Payload:            buffer.MakeWithData(bytes[:read]),
+			Payload:            buffer.MakeWithData(buf[:read]),
 		})
-		//defer pkt.DecRef()
+		config.SPool.Put(buf[:])
-		config.LPool.Put(bytes[:])
 		endpoint.InjectInbound(protocol, pkt)
+		pkt.DecRef()
 		log.Tracef("[TUN-%s] Write to Gvisor IP-Protocol: %s, SRC: %s, DST: %s, Length: %d", layers.IPProtocol(ipProtocol).String(), layers.IPProtocol(ipProtocol).String(), src.String(), dst, read)
 	}
 }

pkg/core/gvisorudpforwarder.go

@@ -2,10 +2,10 @@ package core
 
 import (
 	"context"
-	"errors"
 	"io"
 	"net"
 
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
@@ -17,44 +17,50 @@ import (
 
 func UDPForwarder(s *stack.Stack, ctx context.Context) func(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool {
 	return udp.NewForwarder(s, func(request *udp.ForwarderRequest) {
-		endpointID := request.ID()
+		id := request.ID()
 		log.Debugf("[TUN-UDP] LocalPort: %d, LocalAddress: %s, RemotePort: %d, RemoteAddress %s",
-			endpointID.LocalPort, endpointID.LocalAddress.String(), endpointID.RemotePort, endpointID.RemoteAddress.String(),
+			id.LocalPort, id.LocalAddress.String(), id.RemotePort, id.RemoteAddress.String(),
 		)
+		src := &net.UDPAddr{
+			IP:   id.RemoteAddress.AsSlice(),
+			Port: int(id.RemotePort),
+		}
+		dst := &net.UDPAddr{
+			IP:   id.LocalAddress.AsSlice(),
+			Port: int(id.LocalPort),
+		}
+
 		w := &waiter.Queue{}
 		endpoint, tErr := request.CreateEndpoint(w)
 		if tErr != nil {
-			log.Debugf("[TUN-UDP] Failed to create endpoint: %v", tErr)
+			log.Debugf("[TUN-UDP] Failed to create endpoint to dst: %s: %v", dst.String(), tErr)
 			return
 		}
 
-		// 2, dial proxy
+		// dial dst
-		addr := &net.UDPAddr{
+		remote, err := net.DialUDP("udp", nil, dst)
-			IP:   endpointID.LocalAddress.AsSlice(),
-			Port: int(endpointID.LocalPort),
-		}
-		remote, err := net.DialUDP("udp", nil, addr)
 		if err != nil {
-			log.Errorf("[TUN-UDP] Failed to connect addr %s: %v", addr.String(), err)
+			log.Errorf("[TUN-UDP] Failed to connect dst: %s: %v", dst.String(), err)
 			return
 		}
+
 		conn := gonet.NewUDPConn(w, endpoint)
 		go func() {
 			defer conn.Close()
 			defer remote.Close()
 			errChan := make(chan error, 2)
 			go func() {
-				i := config.LPool.Get().([]byte)[:]
+				buf := config.LPool.Get().([]byte)[:]
-				defer config.LPool.Put(i[:])
+				defer config.LPool.Put(buf[:])
-				written, err2 := io.CopyBuffer(remote, conn, i)
+				written, err2 := io.CopyBuffer(remote, conn, buf)
-				log.Debugf("[TUN-UDP] Write length %d data to remote", written)
+				log.Debugf("[TUN-UDP] Write length %d data from src: %s -> dst: %s", written, src.String(), dst.String())
 				errChan <- err2
 			}()
 			go func() {
-				i := config.LPool.Get().([]byte)[:]
+				buf := config.LPool.Get().([]byte)[:]
-				defer config.LPool.Put(i[:])
+				defer config.LPool.Put(buf[:])
-				written, err2 := io.CopyBuffer(conn, remote, i)
+				written, err2 := io.CopyBuffer(conn, remote, buf)
-				log.Debugf("[TUN-UDP] Read length %d data from remote", written)
+				log.Debugf("[TUN-UDP] Read length %d data from dst: %s -> src: %s", written, dst.String(), src.String())
 				errChan <- err2
 			}()
 			err = <-errChan

pkg/core/gvisorudphandler.go

@@ -102,8 +102,8 @@ func handle(ctx context.Context, tcpConn net.Conn, udpConn *net.UDPConn) {
 	log.Debugf("[TUN-UDP] %s <-> %s", tcpConn.RemoteAddr(), udpConn.LocalAddr())
 	errChan := make(chan error, 2)
 	go func() {
-		b := config.LPool.Get().([]byte)[:]
+		buf := config.LPool.Get().([]byte)[:]
-		defer config.LPool.Put(b[:])
+		defer config.LPool.Put(buf[:])
 
 		for {
 			select {
@@ -118,7 +118,7 @@ func handle(ctx context.Context, tcpConn net.Conn, udpConn *net.UDPConn) {
 				errChan <- err
 				return
 			}
-			dgram, err := readDatagramPacket(tcpConn, b[:])
+			dgram, err := readDatagramPacket(tcpConn, buf[:])
 			if err != nil {
 				log.Errorf("[TUN-UDP] %s -> %s: %v", tcpConn.RemoteAddr(), udpConn.LocalAddr(), err)
 				errChan <- err
@@ -146,8 +146,8 @@ func handle(ctx context.Context, tcpConn net.Conn, udpConn *net.UDPConn) {
 	}()
 
 	go func() {
-		b := config.LPool.Get().([]byte)[:]
+		buf := config.LPool.Get().([]byte)[:]
-		defer config.LPool.Put(b[:])
+		defer config.LPool.Put(buf[:])
 
 		for {
 			select {
@@ -162,7 +162,7 @@ func handle(ctx context.Context, tcpConn net.Conn, udpConn *net.UDPConn) {
 				errChan <- err
 				return
 			}
-			n, _, err := udpConn.ReadFrom(b[:])
+			n, _, err := udpConn.ReadFrom(buf[:])
 			if err != nil {
 				log.Errorf("[TUN-UDP] %s : %s", tcpConn.RemoteAddr(), err)
 				errChan <- err
@@ -181,7 +181,7 @@ func handle(ctx context.Context, tcpConn net.Conn, udpConn *net.UDPConn) {
 				errChan <- err
 				return
 			}
-			dgram := newDatagramPacket(b[:n])
+			dgram := newDatagramPacket(buf[:n])
 			if err = dgram.Write(tcpConn); err != nil {
 				log.Errorf("[TUN-UDP] Error: %s <- %s : %s", tcpConn.RemoteAddr(), dgram.Addr(), err)
 				errChan <- err

pkg/core/tcphandler.go

@@ -78,10 +78,11 @@ func (h *fakeUdpHandler) Handle(ctx context.Context, tcpConn net.Conn) {
 		default:
 		}
 
-		b := config.LPool.Get().([]byte)[:]
+		buf := config.SPool.Get().([]byte)[:]
-		dgram, err := readDatagramPacketServer(tcpConn, b[:])
+		dgram, err := readDatagramPacketServer(tcpConn, buf[:])
 		if err != nil {
 			log.Errorf("[TCP] %s -> %s : %v", tcpConn.RemoteAddr(), tcpConn.LocalAddr(), err)
+			config.SPool.Put(buf[:])
 			return
 		}
 
@@ -89,6 +90,7 @@ func (h *fakeUdpHandler) Handle(ctx context.Context, tcpConn net.Conn) {
 		src, _, err = util.ParseIP(dgram.Data[:dgram.DataLength])
 		if err != nil {
 			log.Errorf("[TCP] Unknown packet")
+			config.SPool.Put(buf[:])
 			continue
 		}
 		value, loaded := h.routeMapTCP.LoadOrStore(src.String(), tcpConn)
@@ -97,7 +99,6 @@ func (h *fakeUdpHandler) Handle(ctx context.Context, tcpConn net.Conn) {
 				h.routeMapTCP.Store(src.String(), tcpConn)
 				log.Debugf("[TCP] Replace route map TCP: %s -> %s-%s", src, tcpConn.LocalAddr(), tcpConn.RemoteAddr())
 			}
-			log.Debugf("[TCP] Find route map TCP: %s -> %s-%s", src, tcpConn.LocalAddr(), tcpConn.RemoteAddr())
 		} else {
 			log.Debugf("[TCP] Add new route map TCP: %s -> %s-%s", src, tcpConn.LocalAddr(), tcpConn.RemoteAddr())
 		}

pkg/core/tunhandler.go

@@ -37,12 +37,12 @@ func NewRouteMap() *RouteMap {
 	}
 }
 
-func (n *RouteMap) LoadOrStore(to net.IP, addr net.Addr) (result net.Addr, load bool) {
+func (n *RouteMap) LoadOrStore(to net.IP, addr net.Addr) (net.Addr, bool) {
 	n.lock.RLock()
-	route, ok := n.routes[to.String()]
+	route, load := n.routes[to.String()]
 	n.lock.RUnlock()
-	if ok && route.String() == addr.String() {
+	if load {
-		return addr, true
+		return route, true
 	}
 
 	n.lock.Lock()
@@ -51,6 +51,12 @@ func (n *RouteMap) LoadOrStore(to net.IP, addr net.Addr) (result net.Addr, load
 	return addr, false
 }
 
+func (n *RouteMap) Store(to net.IP, addr net.Addr) {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+	n.routes[to.String()] = addr
+}
+
 func (n *RouteMap) RouteTo(ip net.IP) net.Addr {
 	n.lock.RLock()
 	defer n.lock.RUnlock()
@@ -90,27 +96,30 @@ type Device struct {
 
 func (d *Device) readFromTun() {
 	for {
-		b := config.LPool.Get().([]byte)[:]
+		buf := config.SPool.Get().([]byte)[:]
-		n, err := d.tun.Read(b[:])
+		n, err := d.tun.Read(buf[:])
 		if err != nil {
+			config.SPool.Put(buf[:])
 			log.Errorf("[TUN] Failed to read from tun: %v", err)
 			util.SafeWrite(d.chExit, err)
 			return
 		}
 		if n == 0 {
 			log.Errorf("[TUN] Read packet length 0")
+			config.SPool.Put(buf[:])
 			continue
 		}
 
-		src, dst, err := util.ParseIP(b[:n])
+		src, dst, err := util.ParseIP(buf[:n])
 		if err != nil {
 			log.Errorf("[TUN] Unknown packet")
+			config.SPool.Put(buf[:])
 			continue
 		}
 
 		log.Debugf("[TUN] SRC: %s --> DST: %s, length: %d", src, dst, n)
 		util.SafeWrite(d.tunInbound, &DataElem{
-			data:   b[:],
+			data:   buf[:],
 			length: n,
 			src:    src,
 			dst:    dst,
@@ -121,7 +130,7 @@ func (d *Device) readFromTun() {
 func (d *Device) writeToTun() {
 	for e := range d.tunOutbound {
 		_, err := d.tun.Write(e.data[:e.length])
-		config.LPool.Put(e.data[:])
+		config.SPool.Put(e.data[:])
 		if err != nil {
 			util.SafeWrite(d.chExit, err)
 			return
@@ -289,27 +298,32 @@ func (p *Peer) sendErr(err error) {
 
 func (p *Peer) readFromConn() {
 	for {
-		b := config.LPool.Get().([]byte)[:]
+		buf := config.SPool.Get().([]byte)[:]
-		n, from, err := p.conn.ReadFrom(b[:])
+		n, from, err := p.conn.ReadFrom(buf[:])
 		if err != nil {
+			config.SPool.Put(buf[:])
 			p.sendErr(err)
 			return
 		}
 
-		src, dst, err := util.ParseIP(b[:n])
+		src, dst, err := util.ParseIP(buf[:n])
 		if err != nil {
+			config.SPool.Put(buf[:])
 			log.Errorf("[TUN] Unknown packet: %v", err)
 			continue
 		}
-		if _, loaded := p.routeMapUDP.LoadOrStore(src, from); loaded {
+		if addr, loaded := p.routeMapUDP.LoadOrStore(src, from); loaded {
-			log.Debugf("[TUN] Find route: %s -> %s", src, from)
+			if addr.String() != from.String() {
+				p.routeMapUDP.Store(src, from)
+				log.Debugf("[TUN] Replace route map UDP: %s -> %s", src, from)
+			}
 		} else {
-			log.Debugf("[TUN] Add new route: %s -> %s", src, from)
+			log.Debugf("[TUN] Add new route map UDP: %s -> %s", src, from)
 		}
 
 		p.connInbound <- &udpElem{
 			from:   from,
-			data:   b[:],
+			data:   buf[:],
 			length: n,
 			src:    src,
 			dst:    dst,
@@ -322,6 +336,7 @@ func (p *Peer) readFromTCPConn() {
 		src, dst, err := util.ParseIP(packet.Data)
 		if err != nil {
 			log.Errorf("[TUN] Unknown packet")
+			config.SPool.Put(packet.Data[:])
 			continue
 		}
 		u := &udpElem{
@@ -338,9 +353,9 @@ func (p *Peer) readFromTCPConn() {
 func (p *Peer) routePeer() {
 	for e := range p.connInbound {
 		if routeToAddr := p.routeMapUDP.RouteTo(e.dst); routeToAddr != nil {
-			log.Debugf("[TCP] Find route: %s -> %s", e.dst, routeToAddr)
+			log.Debugf("[UDP] Find UDP route to dst: %s -> %s", e.dst, routeToAddr)
 			_, err := p.conn.WriteTo(e.data[:e.length], routeToAddr)
-			config.LPool.Put(e.data[:])
+			config.SPool.Put(e.data[:])
 			if err != nil {
 				p.sendErr(err)
 				return
@@ -348,14 +363,15 @@ func (p *Peer) routePeer() {
 		} else if conn, ok := p.routeMapTCP.Load(e.dst.String()); ok {
 			log.Debugf("[TCP] Find TCP route to dst: %s -> %s", e.dst.String(), conn.(net.Conn).RemoteAddr())
 			dgram := newDatagramPacket(e.data[:e.length])
-			if err := dgram.Write(conn.(net.Conn)); err != nil {
+			err := dgram.Write(conn.(net.Conn))
+			config.SPool.Put(e.data[:])
+			if err != nil {
 				log.Errorf("[TCP] udp-tun %s <- %s : %s", conn.(net.Conn).RemoteAddr(), dgram.Addr(), err)
 				p.sendErr(err)
 				return
 			}
-			config.LPool.Put(e.data[:])
 		} else {
-			log.Debugf("[TCP] Not found route to dst: %s, write to TUN device", e.dst.String())
+			log.Debugf("[TUN] Not found route to dst: %s, write to TUN device", e.dst.String())
 			p.tunOutbound <- &DataElem{
 				data:   e.data,
 				length: e.length,
@@ -369,11 +385,11 @@ func (p *Peer) routePeer() {
 func (p *Peer) routeTUN() {
 	for e := range p.tunInbound {
 		if addr := p.routeMapUDP.RouteTo(e.dst); addr != nil {
-			log.Debugf("[TUN] Find route: %s -> %s", e.dst, addr)
+			log.Debugf("[TUN] Find UDP route to dst: %s -> %s", e.dst, addr)
 			_, err := p.conn.WriteTo(e.data[:e.length], addr)
-			config.LPool.Put(e.data[:])
+			config.SPool.Put(e.data[:])
 			if err != nil {
-				log.Debugf("[TUN] Failed to route: %s -> %s", e.dst, addr)
+				log.Debugf("[TUN] Failed wirte to route dst: %s -> %s", e.dst, addr)
 				p.sendErr(err)
 				return
 			}
@@ -381,15 +397,15 @@ func (p *Peer) routeTUN() {
 			log.Debugf("[TUN] Find TCP route to dst: %s -> %s", e.dst.String(), conn.(net.Conn).RemoteAddr())
 			dgram := newDatagramPacket(e.data[:e.length])
 			err := dgram.Write(conn.(net.Conn))
-			config.LPool.Put(e.data[:])
+			config.SPool.Put(e.data[:])
 			if err != nil {
-				log.Errorf("[TUN] udp-tun %s <- %s : %s", conn.(net.Conn).RemoteAddr(), dgram.Addr(), err)
+				log.Errorf("[TUN] Failed to write TCP %s <- %s : %s", conn.(net.Conn).RemoteAddr(), dgram.Addr(), err)
 				p.sendErr(err)
 				return
 			}
 		} else {
-			log.Errorf("[TUN] No route for %s -> %s, drop it", e.src, e.dst)
+			log.Errorf("[TUN] No route for src: %s -> dst: %s, drop it", e.src, e.dst)
-			config.LPool.Put(e.data[:])
+			config.SPool.Put(e.data[:])
 		}
 	}
 }

pkg/core/tunhandlerclient.go

@@ -87,7 +87,7 @@ func transportTunClient(ctx context.Context, tunInbound <-chan *DataElem, tunOut
 				continue
 			}
 			_, err := packetConn.WriteTo(e.data[:e.length], remoteAddr)
-			config.LPool.Put(e.data[:])
+			config.SPool.Put(e.data[:])
 			if err != nil {
 				util.SafeWrite(errChan, errors.Wrap(err, fmt.Sprintf("failed to write packet to remote %s", remoteAddr)))
 				return
@@ -97,13 +97,14 @@ func transportTunClient(ctx context.Context, tunInbound <-chan *DataElem, tunOut
 
 	go func() {
 		for {
-			b := config.LPool.Get().([]byte)[:]
+			buf := config.SPool.Get().([]byte)[:]
-			n, _, err := packetConn.ReadFrom(b[:])
+			n, _, err := packetConn.ReadFrom(buf[:])
 			if err != nil {
+				config.SPool.Put(buf[:])
 				util.SafeWrite(errChan, errors.Wrap(err, fmt.Sprintf("failed to read packet from remote %s", remoteAddr)))
 				return
 			}
-			util.SafeWrite(tunOutbound, &DataElem{data: b[:], length: n})
+			util.SafeWrite(tunOutbound, &DataElem{data: buf[:], length: n})
 		}
 	}()
 
@@ -145,30 +146,35 @@ func (d *ClientDevice) SetTunInboundHandler(handler func(tunInbound <-chan *Data
 
 func (d *ClientDevice) readFromTun() {
 	for {
-		b := config.LPool.Get().([]byte)[:]
+		buf := config.SPool.Get().([]byte)[:]
-		n, err := d.tun.Read(b[:])
+		n, err := d.tun.Read(buf[:])
 		if err != nil {
 			util.SafeWrite(d.chExit, err)
+			config.SPool.Put(buf[:])
 			return
 		}
-		if n != 0 {
+		if n == 0 {
-			// Try to determine network protocol number, default zero.
+			config.SPool.Put(buf[:])
-			var src, dst net.IP
+			continue
-			src, dst, err = util.ParseIP(b[:n])
-			if err != nil {
-				log.Debugf("[TUN-GVISOR] Unknown packet: %v", err)
-				continue
-			}
-			log.Tracef("[TUN-RAW] SRC: %s, DST: %s, Length: %d", src.String(), dst, n)
-			util.SafeWrite(d.tunInbound, NewDataElem(b[:], n, src, dst))
 		}
+
+		// Try to determine network protocol number, default zero.
+		var src, dst net.IP
+		src, dst, err = util.ParseIP(buf[:n])
+		if err != nil {
+			log.Debugf("[TUN-GVISOR] Unknown packet: %v", err)
+			config.SPool.Put(buf[:])
+			continue
+		}
+		log.Tracef("[TUN-RAW] SRC: %s, DST: %s, Length: %d", src.String(), dst, n)
+		util.SafeWrite(d.tunInbound, NewDataElem(buf[:], n, src, dst))
 	}
 }
 
 func (d *ClientDevice) writeToTun() {
 	for e := range d.tunOutbound {
 		_, err := d.tun.Write(e.data[:e.length])
-		config.LPool.Put(e.data[:])
+		config.SPool.Put(e.data[:])
 		if err != nil {
 			util.SafeWrite(d.chExit, err)
 			return

pkg/core/udpovertcp.go

@@ -61,10 +61,10 @@ func readDatagramPacketServer(r io.Reader, b []byte) (*datagramPacket, error) {
 }
 
 func (addr *datagramPacket) Write(w io.Writer) error {
-	b := config.LPool.Get().([]byte)[:]
+	buf := config.MPool.Get().([]byte)[:]
-	defer config.LPool.Put(b[:])
+	defer config.MPool.Put(buf[:])
-	binary.BigEndian.PutUint16(b[:2], uint16(len(addr.Data)))
+	binary.BigEndian.PutUint16(buf[:2], uint16(len(addr.Data)))
-	n := copy(b[2:], addr.Data)
+	n := copy(buf[2:], addr.Data)
-	_, err := w.Write(b[:n+2])
+	_, err := w.Write(buf[:n+2])
 	return err
 }

pkg/handler/connect.go

@@ -367,6 +367,7 @@ func (c *ConnectOptions) startLocalTunServer(ctx context.Context, forwardAddress
 	tunConfig := tun.Config{
 		Addr:   (&net.IPNet{IP: c.localTunIPv4.IP, Mask: net.CIDRMask(32, 32)}).String(),
 		Routes: routes,
+		MTU:    config.DefaultMTU,
 	}
 	if enable, _ := util.IsIPv6Enabled(); enable {
 		tunConfig.Addr6 = (&net.IPNet{IP: c.localTunIPv6.IP, Mask: net.CIDRMask(128, 128)}).String()
@@ -378,7 +379,6 @@ func (c *ConnectOptions) startLocalTunServer(ctx context.Context, forwardAddress
 		log.Errorf("Failed to parse local node %s: %v", localNode, err)
 		return err
 	}
-	node.Values.Add(config.ConfigKubeVPNTransportEngine, string(c.Engine))
 
 	chainNode, err := core.ParseNode(forwardAddress)
 	if err != nil {

pkg/ssh/ssh.go

@@ -263,10 +263,7 @@ func copyStream(ctx context.Context, local net.Conn, remote net.Conn) {
 		if err != nil && !errors.Is(err, net.ErrClosed) && !errors.Is(err, io.EOF) {
 			log.Debugf("Failed to copy remote -> local: %s", err)
 		}
-		select {
+		pkgutil.SafeWrite(chDone, true)
-		case chDone <- true:
-		default:
-		}
 	}()
 
 	// start local -> remote data transfer
@@ -277,10 +274,7 @@ func copyStream(ctx context.Context, local net.Conn, remote net.Conn) {
 		if err != nil && !errors.Is(err, net.ErrClosed) && !errors.Is(err, io.EOF) {
 			log.Debugf("Failed to copy local -> remote: %s", err)
 		}
-		select {
+		pkgutil.SafeWrite(chDone, true)
-		case chDone <- true:
-		default:
-		}
 	}()
 
 	select {

pkg/tun/tun.go

@@ -84,30 +84,30 @@ type tunConn struct {
 
 func (c *tunConn) Read(b []byte) (n int, err error) {
 	offset := device.MessageTransportHeaderSize
-	bytes := config.LPool.Get().([]byte)[:]
+	buf := config.MPool.Get().([]byte)[:]
-	defer config.LPool.Put(bytes[:])
+	defer config.MPool.Put(buf[:])
 
 	var size int
-	size, err = c.ifce.Read(bytes[:], offset)
+	size, err = c.ifce.Read(buf[:], offset)
 	if err != nil {
 		return 0, err
 	}
 	if size == 0 || size > device.MaxSegmentSize-device.MessageTransportHeaderSize {
 		return 0, nil
 	}
-	return copy(b, bytes[offset:offset+size]), nil
+	return copy(b, buf[offset:offset+size]), nil
 }
 
 func (c *tunConn) Write(b []byte) (n int, err error) {
 	if len(b) < device.MessageTransportHeaderSize {
 		return 0, err
 	}
-	bytes := config.LPool.Get().([]byte)[:]
+	buf := config.MPool.Get().([]byte)[:]
-	defer config.LPool.Put(bytes[:])
+	defer config.MPool.Put(buf[:])
 
-	copy(bytes[device.MessageTransportOffsetContent:], b)
+	copy(buf[device.MessageTransportOffsetContent:], b)
 
-	return c.ifce.Write(bytes[:device.MessageTransportOffsetContent+len(b)], device.MessageTransportOffsetContent)
+	return c.ifce.Write(buf[:device.MessageTransportOffsetContent+len(b)], device.MessageTransportOffsetContent)
 }
 
 func (c *tunConn) Close() (err error) {