Be fix goroutine leak (#252)

* fix(gateway):fix goroutine leak

* fix(term):remove useless code

* fix(be):add pprof for debug mode

server/api/term.go

@@ -152,7 +152,6 @@ func (api WebTerminalApi) SshEndpoint(c echo.Context) error {
 		NextTerminal: nextTerminal,
 		Observer:     session.NewObserver(s.ID),
 	}
-	go nextSession.Observer.Start()
 	session.GlobalSessionManager.Add <- nextSession
 
 	termHandler := NewTermHandler(sessionId, isRecording, ws, nextTerminal)

server/app/app.go

@@ -3,6 +3,10 @@ package app
 import (
 	"encoding/json"
 	"fmt"
+	"net/http"
+	_ "net/http/pprof"
+
+	"next-terminal/server/log"
 
 	"next-terminal/server/cli"
 	"next-terminal/server/config"
@@ -104,6 +108,9 @@ func Run() error {
 		if err != nil {
 			return err
 		}
+		go func() {
+			log.Fatal(http.ListenAndServe("localhost:8099", nil))
+		}()
 		fmt.Printf("当前配置为: %v\n", string(jsonBytes))
 	}
 

server/global/gateway/gateway.go

@@ -6,6 +6,9 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"sync"
+
+	"next-terminal/server/log"
 
 	"next-terminal/server/utils"
 
@@ -19,7 +22,7 @@ type Gateway struct {
 	SshClient *ssh.Client
 	Message   string // 失败原因
 
-	tunnels map[string]*Tunnel
+	tunnels *sync.Map
 
 	Add  chan *Tunnel
 	Del  chan string
@@ -34,7 +37,7 @@ func NewGateway(id string, connected bool, message string, client *ssh.Client) *
 		SshClient: client,
 		Add:       make(chan *Tunnel),
 		Del:       make(chan string),
-		tunnels:   map[string]*Tunnel{},
+		tunnels:   new(sync.Map),
 		exit:      make(chan bool, 1),
 	}
 }
@@ -43,12 +46,15 @@ func (g *Gateway) Run() {
 	for {
 		select {
 		case t := <-g.Add:
-			g.tunnels[t.ID] = t
+			g.tunnels.Store(t.ID, t)
+			log.Info("add tunnel: %s", t.ID)
 			go t.Open()
 		case k := <-g.Del:
-			if _, ok := g.tunnels[k]; ok {
+			if val, ok := g.tunnels.Load(k); ok {
-				g.tunnels[k].Close()
+				if vval, vok := val.(*Tunnel); vok {
-				delete(g.tunnels, k)
+					vval.Close()
+					g.tunnels.Delete(k)
+				}
 			}
 		case <-g.exit:
 			return
@@ -57,14 +63,14 @@ func (g *Gateway) Run() {
 }
 
 func (g *Gateway) Close() {
-	if g.SshClient != nil {
+	g.tunnels.Range(func(key, value interface{}) bool {
-		_ = g.SshClient.Close()
+		if val, ok := value.(*Tunnel); ok {
+			val.Close()
 		}
-	for id := range g.tunnels {
+		return true
-		g.CloseSshTunnel(id)
+	})
-	}
-
 	g.exit <- true
+
 }
 
 func (g *Gateway) OpenSshTunnel(id, ip string, port int) (exposedIP string, exposedPort int, err error) {

server/global/gateway/manager.go

@@ -1,7 +1,13 @@
 package gateway
 
+import (
+	"sync"
+
+	"next-terminal/server/log"
+)
+
 type Manager struct {
-	gateways map[string]*Gateway
+	gateways *sync.Map
 
 	Add chan *Gateway
 	Del chan string
@@ -11,7 +17,7 @@ func NewManager() *Manager {
 	return &Manager{
 		Add:      make(chan *Gateway),
 		Del:      make(chan string),
-		gateways: map[string]*Gateway{},
+		gateways: new(sync.Map),
 	}
 }
 
@@ -19,19 +25,26 @@ func (m *Manager) Start() {
 	for {
 		select {
 		case g := <-m.Add:
-			m.gateways[g.ID] = g
+			m.gateways.Store(g.ID, g)
+			log.Info("add gateway: %s", g.ID)
 			go g.Run()
 		case k := <-m.Del:
-			if _, ok := m.gateways[k]; ok {
+			if val, ok := m.gateways.Load(k); ok {
-				m.gateways[k].Close()
+				if vv, vok := val.(*Gateway); vok {
-				delete(m.gateways, k)
+					vv.Close()
+					m.gateways.Delete(k)
+				}
+
 			}
 		}
 	}
 }
 
 func (m Manager) GetById(id string) *Gateway {
-	return m.gateways[id]
+	if val, ok := m.gateways.Load(id); ok {
+		return val.(*Gateway)
+	}
+	return nil
 }
 
 var GlobalGatewayManager *Manager

server/global/gateway/tunnel.go

@@ -19,34 +19,24 @@ type Tunnel struct {
 	ctx        context.Context
 	cancel     context.CancelFunc
 	listener   net.Listener
-	localConnections  []net.Conn
+	err        error
-	remoteConnections []net.Conn
 }
 
 func (r *Tunnel) Open() {
 	localAddr := fmt.Sprintf("%s:%d", r.LocalHost, r.LocalPort)
-
-	go func() {
-		<-r.ctx.Done()
-		_ = r.listener.Close()
-		for i := range r.localConnections {
-			_ = r.localConnections[i].Close()
-		}
-		r.localConnections = nil
-		for i := range r.remoteConnections {
-			_ = r.remoteConnections[i].Close()
-		}
-		r.remoteConnections = nil
-		log.Debugf("SSH 隧道 %v 关闭", localAddr)
-	}()
 	for {
+		select {
+		case <-r.ctx.Done():
+			_ = r.listener.Close()
+			log.Debugf("SSH 隧道 %v 关闭", localAddr)
+			return
+		default:
 			log.Debugf("等待客户端访问 %v", localAddr)
 			localConn, err := r.listener.Accept()
 			if err != nil {
-			log.Debugf("接受连接失败 %v, 退出循环", err.Error())
+				log.Debugf("接受连接失败 %v", err.Error())
-			return
+				continue
 			}
-		r.localConnections = append(r.localConnections, localConn)
 
 			log.Debugf("客户端 %v 连接至 %v", localConn.RemoteAddr().String(), localAddr)
 			remoteAddr := fmt.Sprintf("%s:%d", r.RemoteHost, r.RemotePort)
@@ -54,21 +44,35 @@ func (r *Tunnel) Open() {
 			remoteConn, err := r.Gateway.SshClient.Dial("tcp", remoteAddr)
 			if err != nil {
 				log.Debugf("连接远程主机 %v 失败", remoteAddr)
+				r.err = err
 				return
 			}
-		r.remoteConnections = append(r.remoteConnections, remoteConn)
 
 			log.Debugf("连接远程主机 %v 成功", remoteAddr)
-		go copyConn(localConn, remoteConn)
+			go copyConn(r.ctx, localConn, remoteConn)
-		go copyConn(remoteConn, localConn)
+			go copyConn(r.ctx, remoteConn, localConn)
 			log.Debugf("转发数据 [%v]->[%v]", localAddr, remoteAddr)
 		}
 	}
+}
 
 func (r Tunnel) Close() {
 	r.cancel()
+	err := r.listener.Close()
+	if err != nil {
+		return
+	}
 }
 
-func copyConn(writer, reader net.Conn) {
+func copyConn(ctx context.Context, writer, reader net.Conn) {
 	_, _ = io.Copy(writer, reader)
+	for {
+		select {
+		case <-ctx.Done():
+			_ = writer.Close()
+			_ = reader.Close()
+			return
+		}
+	}
+
 }