perf: use chan to communicate between tcpserver and tun

build/Dockerfile

@@ -16,7 +16,7 @@ ARG BASE=github.com/wencaiwulue/kubevpn
 RUN sed -i s@/security.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list \
     && sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list
 RUN apt-get clean && apt-get update && apt-get install -y wget dnsutils vim curl  \
-    net-tools iptables iputils-ping lsof iproute2 tcpdump binutils traceroute conntrack socat
+    net-tools iptables iputils-ping lsof iproute2 tcpdump binutils traceroute conntrack socat iperf3
 
 ENV TZ=Asia/Shanghai \
     DEBIAN_FRONTEND=noninteractive

build/local.Dockerfile

@@ -8,7 +8,7 @@ FROM ubuntu:latest
 RUN sed -i s@/security.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list \
     && sed -i s@/archive.ubuntu.com/@/mirrors.aliyun.com/@g /etc/apt/sources.list
 RUN apt-get clean && apt-get update && apt-get install -y wget dnsutils vim curl  \
-    net-tools iptables iputils-ping lsof iproute2 tcpdump binutils traceroute conntrack socat
+    net-tools iptables iputils-ping lsof iproute2 tcpdump binutils traceroute conntrack socat iperf3
 
 ENV TZ=Asia/Shanghai \
     DEBIAN_FRONTEND=noninteractive

build/test.Dockerfile

@@ -2,4 +2,6 @@ FROM naison/kubevpn:latest
 
 WORKDIR /app
 
+RUN apt-get clean && apt-get update && apt-get install -y iperf3
+
 COPY bin/kubevpn /usr/local/bin/kubevpn

cmd/kubevpn/cmds/connect.go

@@ -4,6 +4,7 @@ import (
 	"io"
 	defaultlog "log"
 	"os"
+	"runtime"
 	"syscall"
 
 	log "github.com/sirupsen/logrus"
@@ -56,6 +57,7 @@ func CmdConnect(f cmdutil.Factory) *cobra.Command {
 			return handler.SshJump(sshConf, cmd.Flags())
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
+			runtime.GOMAXPROCS(0)
 			if err := connect.InitClient(f); err != nil {
 				return err
 			}

cmd/kubevpn/cmds/serve.go

@@ -1,16 +1,13 @@
 package cmds
 
 import (
-	"context"
 	"math/rand"
-	"os"
-	"os/signal"
-	"syscall"
 	"time"
 
 	"github.com/spf13/cobra"
 	"go.uber.org/automaxprocs/maxprocs"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
+	ctrl "sigs.k8s.io/controller-runtime"
 
 	"github.com/wencaiwulue/kubevpn/pkg/config"
 	"github.com/wencaiwulue/kubevpn/pkg/core"
@@ -37,17 +34,11 @@ func CmdServe(_ cmdutil.Factory) *cobra.Command {
 				return err
 			}
 			defer handler.Final()
-			ctx, cancelFunc := context.WithCancel(context.Background())
-			stopChan := make(chan os.Signal)
-			signal.Notify(stopChan, os.Interrupt, os.Kill, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGKILL /*, syscall.SIGSTOP*/)
-			go func() {
-				<-stopChan
-				cancelFunc()
-			}()
 			servers, err := handler.Parse(*route)
 			if err != nil {
 				return err
 			}
+			ctx := ctrl.SetupSignalHandler()
 			return handler.Run(ctx, servers)
 		},
 	}

go.mod

@@ -195,6 +195,7 @@ require (
 	golang.org/x/term v0.6.0 // indirect
 	golang.org/x/tools v0.6.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
+	gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
 	google.golang.org/api v0.109.0 // indirect
 	google.golang.org/genproto v0.0.0-20230113154510-dbe35b8444a5 // indirect
 	gopkg.in/DataDog/dd-trace-go.v1 v1.47.0 // indirect

go.sum

@@ -397,6 +397,7 @@ github.com/envoyproxy/protoc-gen-validate v0.6.7/go.mod h1:dyJXwwfPK2VSqiB9Klm1J
 github.com/envoyproxy/protoc-gen-validate v0.9.1 h1:PS7VIOgmSVhWUEeZwTe7z7zouA22Cr590PzXKbZHOVY=
 github.com/envoyproxy/protoc-gen-validate v0.9.1/go.mod h1:OKNgG7TCp5pF4d6XftA0++PMirau2/yoOwVac3AbF2w=
 github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5/go.mod h1:a2zkGnVExMxdzMo3M0Hi/3sEU+cWnZpSni0O6/Yb/P0=
+github.com/evanphx/json-patch v0.5.2/go.mod h1:ZWS5hhDbVDyob71nXKNL0+PWn6ToqBHMikGIFbs31qQ=
 github.com/evanphx/json-patch v4.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/evanphx/json-patch v5.6.0+incompatible h1:jBYDEEiFBPxA0v50tFdvOzQQTCvpL6mnFh5mB2/l16U=
 github.com/evanphx/json-patch v5.6.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
@@ -1411,6 +1412,7 @@ golang.zx2c4.com/wireguard v0.0.0-20220920152132-bb719d3a6e2c/go.mod h1:enML0deD
 golang.zx2c4.com/wireguard/windows v0.5.3 h1:On6j2Rpn3OEMXqBq00QEDC7bWSZrPIHKIus8eIuExIE=
 golang.zx2c4.com/wireguard/windows v0.5.3/go.mod h1:9TEe8TJmtwyQebdFwAkEWOPr3prrtqm+REGFifP60hI=
 gomodules.xyz/jsonpatch/v2 v2.2.0 h1:4pT439QV83L+G9FkcCriY6EkpcK6r6bK+A5FBUMI7qY=
+gomodules.xyz/jsonpatch/v2 v2.2.0/go.mod h1:WXp+iVDkoLQqPudfQ9GBlwB2eZ5DKOnjQZCYdOS8GPY=
 google.golang.org/api v0.0.0-20160322025152-9bf6e6e569ff/go.mod h1:4mhQ8q/RsB7i+udVvVy5NUi08OU8ZlA0gRVgrF7VFY0=
 google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
 google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=

pkg/config/config.go

@@ -135,7 +135,7 @@ var (
 	//	network layer ip needs 20 bytes
 	//	transport layer UDP header needs 8 bytes
 	//	UDP over TCP header needs 22 bytes
-	DefaultMTU = 1500 - 20 - 8 - 21
+	DefaultMTU = 65521
 )
 
 var (

pkg/core/route.go

@@ -4,6 +4,7 @@ import (
 	"net"
 	"os"
 	"strings"
+	"sync"
 
 	"github.com/containernetworking/cni/pkg/types"
 	"github.com/pkg/errors"
@@ -15,8 +16,16 @@ import (
 var (
 	// RouteNAT Globe route table for inner ip
 	RouteNAT = NewNAT()
+	// RouteConnNAT map[srcIP]net.Conn
+	RouteConnNAT = &sync.Map{}
+	// Chan tcp connects
+	Chan = make(chan *datagramPacket, MaxSize)
 )
 
+type TCPUDPacket struct {
+	data *datagramPacket
+}
+
 // Route example:
 // -L "tcp://:10800" -L "tun://:8422?net=223.254.0.100/16"
 // -L "tun:/10.233.24.133:8422?net=223.254.0.102/16&route=223.254.0.0/16"

pkg/core/tcphandler.go

@@ -4,11 +4,13 @@ import (
 	"context"
 	"errors"
 	"net"
+	"sync"
 	"time"
 
 	log "github.com/sirupsen/logrus"
 
 	"github.com/wencaiwulue/kubevpn/pkg/config"
+	"github.com/wencaiwulue/kubevpn/pkg/util"
 )
 
 type fakeUDPTunnelConnector struct {
@@ -39,12 +41,15 @@ func (c *fakeUDPTunnelConnector) ConnectContext(ctx context.Context, conn net.Co
 }
 
 type fakeUdpHandler struct {
-	nat *NAT
+	// map[srcIP]net.Conn
+	connNAT *sync.Map
+	ch      chan *datagramPacket
 }
 
 func TCPHandler() Handler {
 	return &fakeUdpHandler{
-		nat: RouteNAT,
+		connNAT: RouteConnNAT,
+		ch:      Chan,
 	}
 }
 
@@ -53,69 +58,49 @@ var Server8422, _ = net.ResolveUDPAddr("udp", "localhost:8422")
 func (h *fakeUdpHandler) Handle(ctx context.Context, tcpConn net.Conn) {
 	defer tcpConn.Close()
 	log.Debugf("[tcpserver] %s -> %s\n", tcpConn.RemoteAddr(), tcpConn.LocalAddr())
-	udpConn, err := net.DialUDP("udp", nil, Server8422)
-	if err != nil {
-		log.Errorf("[tcpserver] udp-tun %s -> %s : %s", tcpConn.RemoteAddr(), udpConn.LocalAddr(), err)
-		return
-	}
-	defer udpConn.Close()
 
 	defer func(addr net.Addr) {
-		n := h.nat.RemoveAddr(addr)
-		log.Debugf("delete addr %s from globle route, deleted count %d", addr, n)
-	}(udpConn.LocalAddr())
-
-	log.Debugf("[tcpserver] udp-tun %s <-> %s", tcpConn.RemoteAddr(), udpConn.LocalAddr())
-	errChan := make(chan error, 2)
-	go func() {
-		b := config.LPool.Get().([]byte)
-		defer config.LPool.Put(b[:])
-
-		for {
-			dgram, err := readDatagramPacket(tcpConn, b[:])
-			if err != nil {
-				log.Debugf("[tcpserver] %s -> 0 : %v", tcpConn.RemoteAddr(), err)
-				errChan <- err
-				return
-			}
-
-			if _, err = udpConn.Write(dgram.Data); err != nil {
-				log.Debugf("[tcpserver] udp-tun %s -> %s : %s", tcpConn.RemoteAddr(), Server8422, err)
-				errChan <- err
-				return
+		var keys []string
+		h.connNAT.Range(func(key, value any) bool {
+			if value.(net.Conn) == tcpConn {
+				keys = append(keys, key.(string))
 			}
-			log.Debugf("[tcpserver] udp-tun %s >>> %s length: %d", tcpConn.RemoteAddr(), Server8422, len(dgram.Data))
+			return true
+		})
+		for _, key := range keys {
+			h.connNAT.Delete(key)
 		}
-	}()
+		log.Debugf("delete conn %s from globle routeConnNAT, deleted count %d", addr, len(keys))
+	}(tcpConn.LocalAddr())
 
-	go func() {
+	var firstIPv4 = true
+	var firstIPv6 = true
+	for {
 		b := config.LPool.Get().([]byte)
-		defer config.LPool.Put(b[:])
-
-		for {
-			n, err := udpConn.Read(b[:])
-			if err != nil {
-				log.Debugf("[tcpserver] %s : %s", tcpConn.RemoteAddr(), err)
-				errChan <- err
-				return
-			}
+		dgram, err := readDatagramPacketServer(tcpConn, b[:])
+		if err != nil {
+			log.Debugf("[tcpserver] %s -> 0 : %v", tcpConn.RemoteAddr(), err)
+			return
+		}
 
-			// pipe from peer to tunnel
-			dgram := newDatagramPacket(b[:n])
-			if err = dgram.Write(tcpConn); err != nil {
-				log.Debugf("[tcpserver] udp-tun %s <- %s : %s", tcpConn.RemoteAddr(), dgram.Addr(), err)
-				errChan <- err
-				return
+		if firstIPv4 || firstIPv6 {
+			var src net.IP
+			bb := dgram.Data[:dgram.DataLength]
+			if util.IsIPv4(bb) {
+				src = net.IPv4(bb[12], bb[13], bb[14], bb[15])
+				firstIPv4 = false
+			} else if util.IsIPv6(bb) {
+				src = bb[8:24]
+				firstIPv6 = false
+			} else {
+				log.Errorf("[tun] unknown packet")
+				continue
 			}
-			log.Debugf("[tcpserver] udp-tun %s <<< %s length: %d", tcpConn.RemoteAddr(), dgram.Addr(), len(dgram.Data))
+			h.connNAT.LoadOrStore(src.String(), tcpConn)
+			log.Debugf("[tun] new routeConnNAT: %s -> %s-%s", src, tcpConn.LocalAddr(), tcpConn.RemoteAddr())
 		}
-	}()
-	err = <-errChan
-	if err != nil {
-		log.Error(err)
+		h.ch <- dgram
 	}
-	log.Debugf("[tcpserver] udp-tun %s >-< %s", tcpConn.RemoteAddr(), udpConn.LocalAddr())
-	return
 }
 
 // fake udp connect over tcp