修改连接池的实现方式

code/client/global/constants.go

@@ -0,0 +1,8 @@
+package global
+
+import "time"
+
+const (
+	ReadTimeout  = time.Second
+	WriteTimeout = time.Second
+)

code/client/main.go

@@ -6,7 +6,11 @@ import (
 	"natpass/code/client/global"
 	"natpass/code/client/pool"
 	"natpass/code/client/tunnel"
+	"natpass/code/network"
+	"net"
 	"os"
+	"strconv"
+	"time"
 
 	"github.com/lwch/daemon"
 	"github.com/lwch/logging"
@@ -55,13 +59,70 @@ func main() {
 	logging.SetSizeRotate(cfg.LogDir, "np-cli", int(cfg.LogSize.Bytes()), cfg.LogRotate, true)
 	defer logging.Flush()
 
-	pl := pool.New(cfg.Links)
+	pl := pool.New(cfg)
 
 	for _, t := range cfg.Tunnels {
-		tn := tunnel.New(t, pl)
-		pl.Add(tn)
-		go tn.Handle()
+		tn := tunnel.New(t)
+		go tn.Handle(pl)
 	}
 
-	pl.Loop(cfg)
+	for i := 0; i < cfg.Links; i++ {
+		go func() {
+			for {
+				conn := pl.Get()
+				for {
+					msg := <-conn.ChanUnknown()
+					if msg == nil {
+						break
+					}
+					var linkID string
+					switch msg.GetXType() {
+					case network.Msg_connect_req:
+						connect(pl, conn, msg.GetFrom(), msg.GetTo(), msg.GetCreq())
+					case network.Msg_connect_rep:
+						linkID = msg.GetCrep().GetId()
+					case network.Msg_disconnect:
+						linkID = msg.GetXDisconnect().GetId()
+					case network.Msg_forward:
+						linkID = msg.GetXData().GetLid()
+					}
+					if len(linkID) > 0 {
+						logging.Error("link of %s not found", linkID)
+						continue
+					}
+				}
+				logging.Info("connection %s exited", conn.ID)
+				time.Sleep(time.Second)
+			}
+		}()
+	}
+
+	select {}
+}
+
+func connect(pool *pool.Pool, conn *pool.Conn, from, to string, req *network.ConnectRequest) {
+	dial := "tcp"
+	if req.GetXType() == network.ConnectRequest_udp {
+		dial = "udp"
+	}
+	link, err := net.Dial(dial, fmt.Sprintf("%s:%d", req.GetAddr(), req.GetPort()))
+	if err != nil {
+		conn.SendConnectError(from, req.GetId(), err.Error())
+		return
+	}
+	host, pt, _ := net.SplitHostPort(link.LocalAddr().String())
+	port, _ := strconv.ParseUint(pt, 10, 16)
+	tn := tunnel.New(global.Tunnel{
+		Name:       req.GetName(),
+		Target:     from,
+		Type:       dial,
+		LocalAddr:  host,
+		LocalPort:  uint16(port),
+		RemoteAddr: req.GetAddr(),
+		RemotePort: uint16(req.GetPort()),
+	})
+	lk := tunnel.NewLink(tn, req.GetId(), from, link, conn)
+	conn.SendConnectOK(from, req.GetId())
+	lk.OnWork <- struct{}{}
+	lk.Forward()
 }

code/client/pool/conn.go

@@ -0,0 +1,130 @@
+package pool
+
+import (
+	"fmt"
+	"natpass/code/client/global"
+	"natpass/code/network"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/lwch/logging"
+)
+
+type Conn struct {
+	sync.RWMutex
+	ID          string
+	parent      *Pool
+	conn        *network.Conn
+	read        map[string]chan *network.Msg // link id => channel
+	unknownRead chan *network.Msg            // read message without link
+	write       chan *network.Msg            // link id => channel
+}
+
+func newConn(parent *Pool, conn *network.Conn) *Conn {
+	ret := &Conn{
+		ID:          fmt.Sprintf("%d", time.Now().UnixNano()),
+		parent:      parent,
+		conn:        conn,
+		read:        make(map[string]chan *network.Msg),
+		unknownRead: make(chan *network.Msg),
+		write:       make(chan *network.Msg),
+	}
+	logging.Info("new connection: %s", ret.ID)
+	go ret.loopRead()
+	go ret.loopWrite()
+	return ret
+}
+
+func (conn *Conn) addLink(id string) {
+	conn.Lock()
+	conn.read[id] = make(chan *network.Msg)
+	conn.Unlock()
+}
+
+// RemoveLink detach read message
+func (conn *Conn) RemoveLink(id string) {
+	conn.Lock()
+	ch := conn.read[id]
+	if ch != nil {
+		close(ch)
+	}
+	delete(conn.read, id)
+	conn.Unlock()
+}
+
+// Close close connection
+func (conn *Conn) Close() {
+	conn.conn.Close()
+	conn.Lock()
+	for id, ch := range conn.read {
+		close(ch)
+		delete(conn.read, id)
+	}
+	conn.Unlock()
+	close(conn.unknownRead)
+	close(conn.write)
+	conn.parent.onClose(conn.ID)
+}
+
+func (conn *Conn) loopRead() {
+	defer conn.Close()
+	for {
+		msg, err := conn.conn.ReadMessage(global.ReadTimeout)
+		if err != nil {
+			if strings.Contains(err.Error(), "i/o timeout") {
+				continue
+			}
+			logging.Error("read message: %v", err)
+			return
+		}
+		var linkID string
+		switch msg.GetXType() {
+		case network.Msg_connect_req:
+			linkID = msg.GetCreq().GetId()
+		case network.Msg_connect_rep:
+			linkID = msg.GetCrep().GetId()
+		case network.Msg_disconnect:
+			linkID = msg.GetXDisconnect().GetId()
+		case network.Msg_forward:
+			linkID = msg.GetXData().GetLid()
+		}
+		conn.RLock()
+		ch := conn.read[linkID]
+		conn.RUnlock()
+		if ch == nil {
+			ch = conn.unknownRead
+		}
+		select {
+		case ch <- msg:
+		case <-time.After(global.ReadTimeout):
+			logging.Error("write read channel for link %s timeouted", linkID)
+			close(ch)
+		}
+	}
+}
+
+func (conn *Conn) loopWrite() {
+	defer conn.Close()
+	for {
+		msg := <-conn.write
+		msg.From = conn.parent.cfg.ID
+		err := conn.conn.WriteMessage(msg, global.WriteTimeout)
+		if err == nil {
+			continue
+		}
+		logging.Error("write message error on %d: %v", conn.ID, err)
+		return
+	}
+}
+
+// ChanRead get read channel from link id
+func (conn *Conn) ChanRead(id string) <-chan *network.Msg {
+	conn.RLock()
+	defer conn.RUnlock()
+	return conn.read[id]
+}
+
+func (conn *Conn) ChanUnknown() <-chan *network.Msg {
+	return conn.unknownRead
+}

code/client/pool/handler.go

@@ -1,112 +0,0 @@
-package pool
-
-import (
-	"fmt"
-	"natpass/code/client/global"
-	"natpass/code/client/tunnel"
-	"natpass/code/network"
-	"net"
-	"strconv"
-	"time"
-
-	"github.com/lwch/logging"
-)
-
-func (p *Pool) handleConnectReq(conn *network.Conn, from, to string, req *network.ConnectRequest) {
-	dial := "tcp"
-	if req.GetXType() == network.ConnectRequest_udp {
-		dial = "udp"
-	}
-	link, err := net.Dial(dial, fmt.Sprintf("%s:%d", req.GetAddr(), req.GetPort()))
-	if err != nil {
-		p.sendConnectError(conn, to, from, req.GetId(), err.Error())
-		return
-	}
-	host, pt, _ := net.SplitHostPort(link.LocalAddr().String())
-	port, _ := strconv.ParseUint(pt, 10, 16)
-	tn := tunnel.New(global.Tunnel{
-		Name:       req.GetName(),
-		Target:     from,
-		Type:       dial,
-		LocalAddr:  host,
-		LocalPort:  uint16(port),
-		RemoteAddr: req.GetAddr(),
-		RemotePort: uint16(req.GetPort()),
-	}, p)
-	tn.NewLink(req.GetId(), from, link)
-	p.Add(tn)
-	p.sendConnectOK(conn, to, from, req.GetId())
-}
-
-func (p *Pool) handleConnectRep(rep *network.ConnectResponse) {
-	id := rep.GetId()
-
-	p.RLock()
-	link := p.links[id]
-	p.RUnlock()
-
-	if link == nil {
-		logging.Error("link %s not found", id)
-		return
-	}
-
-	if rep.GetOk() {
-		link.OnWork <- struct{}{}
-	} else {
-		link.Close()
-	}
-}
-
-func (p *Pool) handleDisconnect(data *network.Disconnect) {
-	id := data.GetId()
-
-	p.RLock()
-	link := p.links[id]
-	p.RUnlock()
-
-	if link != nil {
-		logging.Info("disconnect link %s from remote", id)
-		link.Close()
-	}
-}
-
-func (p *Pool) handleData(data *network.Data) {
-	id := data.GetLid()
-	p.RLock()
-	link := p.links[id]
-	p.RUnlock()
-	if link == nil {
-		logging.Error("link %s not found", id)
-		return
-	}
-	link.WriteData(data.GetData())
-}
-
-func (p *Pool) sendConnectError(conn *network.Conn, from, to, id, m string) {
-	var msg network.Msg
-	msg.From = from
-	msg.To = to
-	msg.XType = network.Msg_connect_rep
-	msg.Payload = &network.Msg_Crep{
-		Crep: &network.ConnectResponse{
-			Id:  id,
-			Ok:  false,
-			Msg: m,
-		},
-	}
-	conn.WriteMessage(&msg, time.Second)
-}
-
-func (c *Pool) sendConnectOK(conn *network.Conn, from, to, id string) {
-	var msg network.Msg
-	msg.From = from
-	msg.To = to
-	msg.XType = network.Msg_connect_rep
-	msg.Payload = &network.Msg_Crep{
-		Crep: &network.ConnectResponse{
-			Id: id,
-			Ok: true,
-		},
-	}
-	conn.WriteMessage(&msg, time.Second)
-}

code/client/pool/pool.go

@@ -1,13 +1,11 @@
 package pool
 
 import (
-	"context"
 	"crypto/tls"
 	"fmt"
+	"math/rand"
 	"natpass/code/client/global"
-	"natpass/code/client/tunnel"
 	"natpass/code/network"
-	"strings"
 	"sync"
 	"time"
 
@@ -15,134 +13,79 @@ import (
 	"github.com/lwch/runtime"
 )
 
-type connectData struct {
-	to   string
-	id   string
-	name string
-	tp   network.ConnectRequestType
-	addr string
-	port uint16
-}
-
-type disconnectData struct {
-	to string
-	id string
-}
-
-type forwardData struct {
-	to   string
-	id   string
-	data []byte
-}
-
 // Pool connection pool
 type Pool struct {
 	sync.RWMutex
-	count           int
-	writeConnect    chan connectData
-	writeDisconnect chan disconnectData
-	writeForward    chan forwardData
-	tunnels         map[string]*tunnel.Tunnel // tunnel name => tunnel
-	links           map[string]*tunnel.Link   // link id => link
+	cfg   *global.Configure
+	conns map[string]*Conn
+	count int
+	idx   int
 }
 
 // New create connection pool
-func New(count int) *Pool {
+func New(cfg *global.Configure) *Pool {
 	return &Pool{
-		count:           count,
-		writeConnect:    make(chan connectData),
-		writeDisconnect: make(chan disconnectData),
-		writeForward:    make(chan forwardData),
-		tunnels:         make(map[string]*tunnel.Tunnel),
-		links:           make(map[string]*tunnel.Link),
+		cfg:   cfg,
+		conns: make(map[string]*Conn, cfg.Links),
+		count: cfg.Links,
+		idx:   0,
 	}
 }
 
-// Loop main loop
-func (p *Pool) Loop(cfg *global.Configure) {
-	for i := 0; i < p.count; i++ {
-		go func(i int) {
-			for {
-				p.connect(cfg, i)
-				time.Sleep(time.Second)
-			}
-		}(i)
+func (p *Pool) getConns() []*Conn {
+	ret := make([]*Conn, len(p.conns))
+	p.RLock()
+	for _, conn := range p.conns {
+		ret = append(ret, conn)
 	}
-	select {}
+	p.RUnlock()
+	return ret
 }
 
-// LinkClose on close link
-func (p *Pool) LinkClose(name, id string) {
-	p.Lock()
-	defer p.Unlock()
-	if tunnel, ok := p.tunnels[name]; ok {
-		if len(tunnel.GetLinks()) == 0 {
-			delete(p.tunnels, name)
+// Get get connection
+func (p *Pool) Get(id ...string) *Conn {
+	conns := p.getConns()
+	if len(conns) >= p.count {
+		p.Lock()
+		conn := conns[p.idx%len(conns)]
+		p.idx++
+		p.Unlock()
+		if len(id) > 0 {
+			conn.addLink(id[0])
 		}
+		return conn
 	}
-	delete(p.links, id)
+
+	conn := p.connect()
+	if conn == nil {
+		return nil
+	}
+	c := newConn(p, conn)
+	if len(id) > 0 {
+		c.addLink(id[0])
+	}
+
+	p.Lock()
+	p.conns[c.ID] = c
+	p.idx++
+	p.Unlock()
+	return c
 }
 
-func (p *Pool) connect(cfg *global.Configure, idx int) {
+func (p *Pool) connect() *network.Conn {
 	defer func() {
 		if err := recover(); err != nil {
 			logging.Error("connect error: %v", err)
 		}
 	}()
-	conn, err := tls.Dial("tcp", cfg.Server, nil)
+	conn, err := tls.Dial("tcp", p.cfg.Server, nil)
 	runtime.Assert(err)
 	c := network.NewConn(conn)
 	defer c.Close()
-	err = p.writeHandshake(c, cfg, idx)
+	err = p.writeHandshake(c, p.cfg, rand.Int())
 	runtime.Assert(err)
-	logging.Info("%s connected", cfg.Server)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	go p.send(ctx, c, fmt.Sprintf("%s-%d", cfg.ID, idx))
-	for {
-		msg, err := c.ReadMessage(time.Second)
-		if err != nil {
-			if strings.Contains(err.Error(), "i/o timeout") {
-				continue
-			}
-			logging.Error("read message: %v", err)
-			return
-		}
-		from := msg.GetFrom()
-		if len(from) > 0 {
-			n := strings.LastIndex(from, "-")
-			if n != -1 {
-				from = from[:n]
-			}
-			msg.From = from
-		}
-		switch msg.GetXType() {
-		case network.Msg_connect_req:
-			p.handleConnectReq(c, msg.GetFrom(), msg.GetTo(), msg.GetCreq())
-		case network.Msg_connect_rep:
-			p.handleConnectRep(msg.GetCrep())
-		case network.Msg_disconnect:
-			p.handleDisconnect(msg.GetXDisconnect())
-		case network.Msg_forward:
-			p.handleData(msg.GetXData())
-		}
-	}
-}
-
-// Add add tunnel
-func (p *Pool) Add(tunnel *tunnel.Tunnel) {
-	p.Lock()
-	defer p.Unlock()
-	p.tunnels[tunnel.Name] = tunnel
-	for _, link := range tunnel.GetLinks() {
-		p.links[link.ID] = link
-	}
-}
-
-// AddLink add link
-func (p *Pool) AddLink(link *tunnel.Link) {
-	p.links[link.ID] = link
+	logging.Info("%s connected", p.cfg.Server)
+	return c
 }
 
 func (p *Pool) writeHandshake(conn *network.Conn, cfg *global.Configure, idx int) error {
@@ -158,43 +101,8 @@ func (p *Pool) writeHandshake(conn *network.Conn, cfg *global.Configure, idx int
 	return conn.WriteMessage(&msg, 5*time.Second)
 }
 
-func (p *Pool) send(ctx context.Context, conn *network.Conn, from string) {
-	for {
-		var msg network.Msg
-		msg.From = from
-		select {
-		case m := <-p.writeConnect:
-			msg.XType = network.Msg_connect_req
-			msg.To = m.to
-			msg.Payload = &network.Msg_Creq{
-				Creq: &network.ConnectRequest{
-					Id:    m.id,
-					Name:  m.name,
-					XType: m.tp,
-					Addr:  m.addr,
-					Port:  uint32(m.port),
-				},
-			}
-		case m := <-p.writeDisconnect:
-			msg.XType = network.Msg_disconnect
-			msg.To = m.to
-			msg.Payload = &network.Msg_XDisconnect{
-				XDisconnect: &network.Disconnect{
-					Id: m.id,
-				},
-			}
-		case m := <-p.writeForward:
-			msg.XType = network.Msg_forward
-			msg.To = m.to
-			msg.Payload = &network.Msg_XData{
-				XData: &network.Data{
-					Lid:  m.id,
-					Data: m.data,
-				},
-			}
-		case <-ctx.Done():
-			return
-		}
-		conn.WriteMessage(&msg, time.Second)
-	}
+func (p *Pool) onClose(id string) {
+	p.Lock()
+	delete(p.conns, id)
+	p.Unlock()
 }

code/client/pool/send.go

@@ -3,39 +3,86 @@ package pool
 import (
 	"natpass/code/client/global"
 	"natpass/code/network"
+	"time"
 )
 
-// SendConnect send connect request message
-func (p *Pool) SendConnect(id string, t global.Tunnel) {
+// SendConnectReq send connect request message
+func (conn *Conn) SendConnectReq(id string, cfg global.Tunnel) {
 	tp := network.ConnectRequest_tcp
-	if t.Type != "tcp" {
+	if cfg.Type != "tcp" {
 		tp = network.ConnectRequest_udp
 	}
-	p.writeConnect <- connectData{
-		to:   t.Target,
-		id:   id,
-		name: t.Name,
-		tp:   tp,
-		addr: t.RemoteAddr,
-		port: t.RemotePort,
+	var msg network.Msg
+	msg.To = cfg.Target
+	msg.XType = network.Msg_connect_req
+	msg.Payload = &network.Msg_Creq{
+		Creq: &network.ConnectRequest{
+			Id:    id,
+			Name:  cfg.Name,
+			XType: tp,
+			Addr:  cfg.RemoteAddr,
+			Port:  uint32(cfg.RemotePort),
+		},
+	}
+	select {
+	case conn.write <- &msg:
+	case <-time.After(global.WriteTimeout):
 	}
 }
 
-// SendDisconnect send disconnect message
-func (p *Pool) SendDisconnect(id, to string) {
-	p.writeDisconnect <- disconnectData{
-		to: to,
-		id: id,
+// SendConnectError send connect error response message
+func (conn *Conn) SendConnectError(to, id, info string) {
+	var msg network.Msg
+	msg.To = to
+	msg.XType = network.Msg_connect_rep
+	msg.Payload = &network.Msg_Crep{
+		Crep: &network.ConnectResponse{
+			Id:  id,
+			Ok:  false,
+			Msg: info,
+		},
+	}
+	select {
+	case conn.write <- &msg:
+	case <-time.After(global.WriteTimeout):
 	}
 }
 
-// SendData send forward data
-func (p *Pool) SendData(id, to string, data []byte) {
-	dup := make([]byte, len(data))
-	copy(dup, data)
-	p.writeForward <- forwardData{
-		to:   to,
-		id:   id,
-		data: dup,
+// SendConnectOK send connect success response message
+func (conn *Conn) SendConnectOK(to, id string) {
+	var msg network.Msg
+	msg.To = to
+	msg.XType = network.Msg_connect_rep
+	msg.Payload = &network.Msg_Crep{
+		Crep: &network.ConnectResponse{
+			Id: id,
+			Ok: true,
+		},
+	}
+	select {
+	case conn.write <- &msg:
+	case <-time.After(global.WriteTimeout):
+	}
+}
+
+// SendData forward data
+func (conn *Conn) SendData(id, target string, data []byte) {
+	dup := func(data []byte) []byte {
+		ret := make([]byte, len(data))
+		copy(ret, data)
+		return ret
+	}
+	var msg network.Msg
+	msg.To = target
+	msg.XType = network.Msg_forward
+	msg.Payload = &network.Msg_XData{
+		XData: &network.Data{
+			Lid:  id,
+			Data: dup(data),
+		},
+	}
+	select {
+	case conn.write <- &msg:
+	case <-time.After(global.WriteTimeout):
 	}
 }

code/client/tunnel/link.go

@@ -3,70 +3,83 @@ package tunnel
 import (
 	"bytes"
 	"io"
+	"natpass/code/client/pool"
+	"natpass/code/network"
 	"net"
 
 	"github.com/lwch/logging"
 )
 
 type Link struct {
-	tunnel *Tunnel
-	ID     string // link id
-	target string // remote client id
-	conn   net.Conn
+	parent *Tunnel
+	id     string // link id
+	target string // target id
+	local  net.Conn
+	remote *pool.Conn
 	OnWork chan struct{}
-	closed bool
 }
 
-func newLink(id, target string, tunnel *Tunnel, conn net.Conn) *Link {
-	logging.Info("create link %s: %s", tunnel.Name, id)
+func NewLink(parent *Tunnel, id, target string, local net.Conn, remote *pool.Conn) *Link {
 	return &Link{
-		tunnel: tunnel,
-		ID:     id,
+		parent: parent,
+		id:     id,
 		target: target,
-		conn:   conn,
+		local:  local,
+		remote: remote,
 		OnWork: make(chan struct{}),
-		closed: false,
 	}
 }
 
-// Close close link
-func (link *Link) Close() {
-	if link.closed {
-		return
-	}
-	logging.Info("disconnect tunnel %s on link %s", link.tunnel.Name, link.ID)
-	link.closed = true
-	err := link.conn.Close()
-	if err == nil {
-		link.tunnel.super.SendDisconnect(link.ID, link.target)
-	}
-	link.tunnel.Close(link)
+func (link *Link) close() {
+	logging.Info("close link %s on tunnel %s", link.id, link.parent.Name)
+	link.local.Close()
+	link.remote.RemoveLink(link.id)
 }
 
-func (link *Link) loop() {
-	defer link.Close()
+// Forward forward data
+func (link *Link) Forward() {
+	go link.remoteRead()
+	go link.localRead()
+}
+
+func (link *Link) remoteRead() {
+	defer link.close()
+	ch := link.remote.ChanRead(link.id)
+	for {
+		msg := <-ch
+		switch msg.GetXType() {
+		case network.Msg_forward:
+			_, err := io.Copy(link.local, bytes.NewReader(msg.GetXData().GetData()))
+			if err != nil {
+				logging.Error("write data on tunnel %s link %s failed, err=%v", link.parent.Name, link.id, err)
+				return
+			}
+		case network.Msg_connect_rep:
+			if msg.GetCrep().GetOk() {
+				link.OnWork <- struct{}{}
+				continue
+			}
+			logging.Error("create link %s on tunnel %s failed, err=%s",
+				link.id, link.parent.Name, msg.GetCrep().GetMsg())
+			return
+		}
+	}
+}
+
+func (link *Link) localRead() {
+	defer link.close()
 	<-link.OnWork
 	buf := make([]byte, 32*1024)
 	for {
-		n, err := link.conn.Read(buf)
+		n, err := link.local.Read(buf)
 		if err != nil {
-			logging.Error("read data on tunnel %s link %s failed, err=%v", link.tunnel.Name, link.ID, err)
+			logging.Error("read data on tunnel %s link %s failed, err=%v", link.parent.Name, link.id, err)
 			return
 		}
 		if n == 0 {
 			continue
 		}
-		logging.Debug("link %s on tunnel %s read from local %d bytes", link.ID, link.tunnel.Name, n)
-		link.tunnel.super.SendData(link.ID, link.target, buf[:n])
+		logging.Debug("link %s on tunnel %s read from local %d bytes", link.id, link.parent.Name, n)
+		link.remote.SendData(link.id, link.target, buf[:n])
 	}
 }
-
-// WriteData write data from remote
-func (link *Link) WriteData(data []byte) error {
-	logging.Debug("link %s on tunnel %s write from remote %d bytes", link.ID, link.tunnel.Name, len(data))
-	_, err := io.Copy(link.conn, bytes.NewReader(data))
-	if err != nil {
-		logging.Error("write data on tunnel %s link %s failed, err=%v", link.tunnel.Name, link.ID, err)
-	}
-	return err
-}

code/client/tunnel/tunnel.go

@@ -2,61 +2,37 @@ package tunnel
 
 import (
 	"natpass/code/client/global"
+	"natpass/code/client/pool"
 	"net"
-	"sync"
 
 	"github.com/lwch/logging"
 	"github.com/lwch/runtime"
 )
 
-type Super interface {
-	LinkClose(string, string) // tunnel name, link id
-	AddLink(*Link)
-	SendConnect(string, global.Tunnel)
-	SendDisconnect(string, string)
-	SendData(id, to string, data []byte)
-}
-
-// Tunnel tunnel
 type Tunnel struct {
-	sync.RWMutex
-	super Super
-	Name  string
-	cfg   global.Tunnel
-	links map[string]*Link
+	Name string
+	cfg  global.Tunnel
 }
 
-// New create tunnel
-func New(cfg global.Tunnel, super Super) *Tunnel {
+// New new tunnel
+func New(cfg global.Tunnel) *Tunnel {
 	return &Tunnel{
-		super: super,
-		Name:  cfg.Name,
-		cfg:   cfg,
-		links: make(map[string]*Link),
+		Name: cfg.Name,
+		cfg:  cfg,
 	}
 }
 
-func (tunnel *Tunnel) NewLink(id, target string, conn net.Conn) {
-	link := newLink(id, target, tunnel, conn)
-	tunnel.Lock()
-	tunnel.links[link.ID] = link
-	tunnel.Unlock()
-	go link.loop()
-	link.OnWork <- struct{}{}
-}
-
-// Handle tunnel handler
-func (tunnel *Tunnel) Handle() {
+// Handle handle tunnel
+func (tunnel *Tunnel) Handle(pool *pool.Pool) {
 	if tunnel.cfg.Type == "tcp" {
-		tunnel.handleTcp()
+		tunnel.handleTcp(pool)
 	} else {
 		// TODO
 		func() {}()
 	}
 }
 
-// handleTcp local listen to tcp tunnel
-func (tunnel *Tunnel) handleTcp() {
+func (tunnel *Tunnel) handleTcp(pool *pool.Pool) {
 	defer func() {
 		if err := recover(); err != nil {
 			logging.Error("close tcp tunnel: %s, err=%v", tunnel.cfg.Name, err)
@@ -82,34 +58,14 @@ func (tunnel *Tunnel) handleTcp() {
 			continue
 		}
 
-		link := newLink(id, tunnel.cfg.Target, tunnel, conn)
-		tunnel.super.SendConnect(link.ID, tunnel.cfg)
-
-		tunnel.super.AddLink(link)
-		tunnel.Lock()
-		tunnel.links[link.ID] = link
-		tunnel.Unlock()
-
-		go link.loop()
+		remote := pool.Get(id)
+		if remote == nil {
+			conn.Close()
+			logging.Error("no connection available")
+			continue
+		}
+		link := NewLink(tunnel, id, tunnel.cfg.Target, conn, remote)
+		remote.SendConnectReq(id, tunnel.cfg)
+		link.Forward()
 	}
 }
-
-// Close close link
-func (tunnel *Tunnel) Close(link *Link) {
-	link.Close()
-	tunnel.Lock()
-	delete(tunnel.links, link.ID)
-	tunnel.Unlock()
-	tunnel.super.LinkClose(tunnel.Name, link.ID)
-}
-
-// GetLinks get tunnel links
-func (tunnel *Tunnel) GetLinks() []*Link {
-	ret := make([]*Link, 0, len(tunnel.links))
-	tunnel.RLock()
-	for _, l := range tunnel.links {
-		ret = append(ret, l)
-	}
-	tunnel.RUnlock()
-	return ret
-}