优化udp代理

config.go

@@ -124,7 +124,7 @@ func initConfig() (err error) {
 	httpArgs.Debug = isDebug
 	//########tcp#########
 	tcp := app.Command("tcp", "proxy on tcp mode")
-	tcpArgs.Parent = tcp.Flag("parent", "parent address, such as: \"23.32.32.19:28008\"").Default("[]").Short('P').String()
+	tcpArgs.Parent = tcp.Flag("parent", "parent address, such as: \"23.32.32.19:28008\"").Default("").Short('P').String()
 	tcpArgs.CertFile = tcp.Flag("cert", "cert file for tls").Short('C').Default("proxy.crt").String()
 	tcpArgs.KeyFile = tcp.Flag("key", "key file for tls").Short('K').Default("proxy.key").String()
 	tcpArgs.Timeout = tcp.Flag("timeout", "tcp timeout milliseconds when connect to real server or parent proxy").Short('e').Default("2000").Int()
@@ -136,7 +136,7 @@ func initConfig() (err error) {
 
 	//########udp#########
 	udp := app.Command("udp", "proxy on udp mode")
-	udpArgs.Parent = udp.Flag("parent", "parent address, such as: \"23.32.32.19:28008\"").Default("[]").Short('P').String()
+	udpArgs.Parent = udp.Flag("parent", "parent address, such as: \"23.32.32.19:28008\"").Default("").Short('P').String()
 	udpArgs.CertFile = udp.Flag("cert", "cert file for tls").Short('C').Default("proxy.crt").String()
 	udpArgs.KeyFile = udp.Flag("key", "key file for tls").Short('K').Default("proxy.key").String()
 	udpArgs.Timeout = udp.Flag("timeout", "tcp timeout milliseconds when connect to parent proxy").Short('t').Default("2000").Int()

core/lib/mapx/map.go

@@ -192,13 +192,12 @@ func snapshot(m ConcurrentMap) (chans []chan Tuple) {
 	wg.Add(SHARD_COUNT)
 	// Foreach shard.
 	for index, shard := range m {
-		go func() {
+		go func(index int, shard *ConcurrentMapShared) {
 			defer func() {
 				if e := recover(); e != nil {
 					fmt.Printf("crashed:%s", string(debug.Stack()))
 				}
 			}()
-			func(index int, shard *ConcurrentMapShared) {
 			// Foreach key, value pair.
 			shard.RLock()
 			chans[index] = make(chan Tuple, len(shard.items))
@@ -209,7 +208,6 @@ func snapshot(m ConcurrentMap) (chans []chan Tuple) {
 			shard.RUnlock()
 			close(chans[index])
 		}(index, shard)
-		}()
 	}
 	wg.Wait()
 	return chans

sdk/android-ios/sdk.go

@@ -143,7 +143,7 @@ func StartWithLog(serviceID, serviceArgsStr string, loggerCallback LogCallback)
 	httpArgs.Debug = debug
 	//########tcp#########
 	tcp := app.Command("tcp", "proxy on tcp mode")
-	tcpArgs.Parent = tcp.Flag("parent", "parent address, such as: \"23.32.32.19:28008\"").Default("[]").Short('P').String()
+	tcpArgs.Parent = tcp.Flag("parent", "parent address, such as: \"23.32.32.19:28008\"").Default("").Short('P').String()
 	tcpArgs.CertFile = tcp.Flag("cert", "cert file for tls").Short('C').Default("proxy.crt").String()
 	tcpArgs.KeyFile = tcp.Flag("key", "key file for tls").Short('K').Default("proxy.key").String()
 	tcpArgs.Timeout = tcp.Flag("timeout", "tcp timeout milliseconds when connect to real server or parent proxy").Short('e').Default("2000").Int()
@@ -155,7 +155,7 @@ func StartWithLog(serviceID, serviceArgsStr string, loggerCallback LogCallback)
 
 	//########udp#########
 	udp := app.Command("udp", "proxy on udp mode")
-	udpArgs.Parent = udp.Flag("parent", "parent address, such as: \"23.32.32.19:28008\"").Default("[]").Short('P').String()
+	udpArgs.Parent = udp.Flag("parent", "parent address, such as: \"23.32.32.19:28008\"").Default("").Short('P').String()
 	udpArgs.CertFile = udp.Flag("cert", "cert file for tls").Short('C').Default("proxy.crt").String()
 	udpArgs.KeyFile = udp.Flag("key", "key file for tls").Short('K').Default("proxy.key").String()
 	udpArgs.Timeout = udp.Flag("timeout", "tcp timeout milliseconds when connect to parent proxy").Short('t').Default("2000").Int()

services/mux/mux_server.go

@@ -182,11 +182,6 @@ func NewMuxServer() services.Service {
 	}
 }
 
-type MuxUDPPacketItem struct {
-	packet    *[]byte
-	localAddr *net.UDPAddr
-	srcAddr   *net.UDPAddr
-}
 type MuxUDPConnItem struct {
 	conn      *net.Conn
 	touchtime int64

services/tcp/tcp.go

@@ -1,13 +1,13 @@
 package tcp
 
 import (
-	"bufio"
 	"crypto/tls"
 	"fmt"
 	"io"
 	logger "log"
 	"net"
 	"runtime/debug"
+	"strings"
 	"time"
 
 	"github.com/snail007/goproxy/services"
@@ -33,7 +33,15 @@ type TCPArgs struct {
 	KCP                 kcpcfg.KCPConfigArgs
 	Jumper              *string
 }
-
+type UDPConnItem struct {
+	conn      *net.Conn
+	isActive  bool
+	touchtime int64
+	srcAddr   *net.UDPAddr
+	localAddr *net.UDPAddr
+	udpConn   *net.UDPConn
+	connid    string
+}
 type TCP struct {
 	cfg       TCPArgs
 	sc        *utils.ServerChannel
@@ -41,6 +49,7 @@ type TCP struct {
 	userConns mapx.ConcurrentMap
 	log       *logger.Logger
 	jumper    *jumper.Jumper
+	udpConns  mapx.ConcurrentMap
 }
 
 func NewTCP() services.Service {
@@ -48,6 +57,7 @@ func NewTCP() services.Service {
 		cfg:       TCPArgs{},
 		isStop:    false,
 		userConns: mapx.NewConcurrentMap(),
+		udpConns:  mapx.NewConcurrentMap(),
 	}
 }
 func (s *TCP) CheckArgs() (err error) {
@@ -81,7 +91,7 @@ func (s *TCP) CheckArgs() (err error) {
 	return
 }
 func (s *TCP) InitService() (err error) {
-
+	s.UDPGCDeamon()
 	return
 }
 func (s *TCP) StopService() {
@@ -158,12 +168,14 @@ func (s *TCP) callback(inConn net.Conn) {
 	case "tls":
 		err = s.OutToTCP(&inConn)
 	case "udp":
-		err = s.OutToUDP(&inConn)
+		s.OutToUDP(&inConn)
 	default:
 		err = fmt.Errorf("unkown parent type %s", *s.cfg.ParentType)
 	}
 	if err != nil {
+		if !utils.IsNetClosedErr(err) {
 			s.log.Printf("connect to %s parent %s fail, ERR:%s", *s.cfg.ParentType, lbAddr, err)
+		}
 		utils.CloseConn(&inConn)
 	}
 }
@@ -192,56 +204,136 @@ func (s *TCP) OutToTCP(inConn *net.Conn) (err error) {
 	return
 }
 func (s *TCP) OutToUDP(inConn *net.Conn) (err error) {
-	s.log.Printf("conn created , remote : %s ", (*inConn).RemoteAddr())
+	var item *UDPConnItem
+	var body []byte
+	srcAddr := ""
+	defer func() {
+		if item != nil {
+			(*(*item).conn).Close()
+			(*item).udpConn.Close()
+			s.udpConns.Remove(srcAddr)
+			(*inConn).Close()
+		}
+	}()
 	for {
 		if s.isStop {
+			return
+		}
+		var srcAddr string
+		srcAddr, body, err = utils.ReadUDPPacket(*inConn)
+		if err != nil {
+			if strings.Contains(err.Error(), "n != int(") {
+				continue
+			}
+			if !utils.IsNetDeadlineErr(err) && err != io.EOF && !utils.IsNetClosedErr(err) {
+				s.log.Printf("udp packet revecived from client fail, err: %s", err)
+			}
+			return
+		}
+		localAddr := *s.cfg.Parent
+		if v, ok := s.udpConns.Get(srcAddr); !ok {
+			_srcAddr, _ := net.ResolveUDPAddr("udp", srcAddr)
+			zeroAddr, _ := net.ResolveUDPAddr("udp", ":")
+			_localAddr, _ := net.ResolveUDPAddr("udp", localAddr)
+			var c *net.UDPConn
+			c, err = net.DialUDP("udp", zeroAddr, _localAddr)
+			if err != nil {
+				s.log.Printf("create local udp conn fail, err : %s", err)
 				(*inConn).Close()
 				return
 			}
-		srcAddr, body, err := utils.ReadUDPPacket(bufio.NewReader(*inConn))
+			item = &UDPConnItem{
-		if err == io.EOF || err == io.ErrUnexpectedEOF {
+				conn:      inConn,
-			//s.log.Printf("connection %s released", srcAddr)
+				srcAddr:   _srcAddr,
-			utils.CloseConn(inConn)
+				localAddr: _localAddr,
-			break
+				udpConn:   c,
 			}
-		//log.Debugf("udp packet revecived:%s,%v", srcAddr, body)
+			s.udpConns.Set(srcAddr, item)
-		dstAddr, err := net.ResolveUDPAddr("udp", *s.cfg.Parent)
+			s.UDPRevecive(srcAddr)
+		} else {
+			item = v.(*UDPConnItem)
+		}
+		(*item).touchtime = time.Now().Unix()
+		go (*item).udpConn.Write(body)
+	}
+}
+func (s *TCP) UDPRevecive(key string) {
+	go func() {
+		defer func() {
+			if e := recover(); e != nil {
+				fmt.Printf("crashed:%s", string(debug.Stack()))
+			}
+		}()
+		s.log.Printf("udp conn %s connected", key)
+		v, ok := s.udpConns.Get(key)
+		if !ok {
+			s.log.Printf("[warn] udp conn not exists for %s", key)
+			return
+		}
+		cui := v.(*UDPConnItem)
+		buf := utils.LeakyBuffer.Get()
+		defer func() {
+			utils.LeakyBuffer.Put(buf)
+			(*cui.conn).Close()
+			cui.udpConn.Close()
+			s.udpConns.Remove(key)
+			s.log.Printf("udp conn %s released", key)
+		}()
+		for {
+			n, err := cui.udpConn.Read(buf)
 			if err != nil {
-			s.log.Printf("can't resolve address: %s", err)
+				if !utils.IsNetClosedErr(err) {
-			utils.CloseConn(inConn)
+					s.log.Printf("udp conn read udp packet fail , err: %s ", err)
-			break
-		}
-		clientSrcAddr := &net.UDPAddr{IP: net.IPv4zero, Port: 0}
-		conn, err := net.DialUDP("udp", clientSrcAddr, dstAddr)
-		if err != nil {
-			s.log.Printf("connect to udp %s fail,ERR:%s", dstAddr.String(), err)
-			continue
-		}
-		conn.SetDeadline(time.Now().Add(time.Millisecond * time.Duration(*s.cfg.Timeout)))
-		_, err = conn.Write(body)
-		if err != nil {
-			s.log.Printf("send udp packet to %s fail,ERR:%s", dstAddr.String(), err)
-			continue
-		}
-		//log.Debugf("send udp packet to %s success", dstAddr.String())
-		buf := make([]byte, 512)
-		len, _, err := conn.ReadFromUDP(buf)
-		if err != nil {
-			s.log.Printf("read udp response from %s fail ,ERR:%s", dstAddr.String(), err)
-			continue
-		}
-		respBody := buf[0:len]
-		//log.Debugf("revecived udp packet from %s , %v", dstAddr.String(), respBody)
-		_, err = (*inConn).Write(utils.UDPPacket(srcAddr, respBody))
-		if err != nil {
-			s.log.Printf("send udp response fail ,ERR:%s", err)
-			utils.CloseConn(inConn)
-			break
-		}
-		//s.log.Printf("send udp response success ,from:%s", dstAddr.String())
 				}
 				return
-
+			}
+			cui.touchtime = time.Now().Unix()
+			go func() {
+				defer func() {
+					if e := recover(); e != nil {
+						fmt.Printf("crashed:%s", string(debug.Stack()))
+					}
+				}()
+				(*cui.conn).SetWriteDeadline(time.Now().Add(time.Millisecond * time.Duration(*s.cfg.Timeout)))
+				_, err = (*cui.conn).Write(utils.UDPPacket(cui.srcAddr.String(), buf[:n]))
+				(*cui.conn).SetWriteDeadline(time.Time{})
+				if err != nil {
+					cui.udpConn.Close()
+					return
+				}
+			}()
+		}
+	}()
+}
+func (s *TCP) UDPGCDeamon() {
+	gctime := int64(30)
+	go func() {
+		defer func() {
+			if e := recover(); e != nil {
+				fmt.Printf("crashed:%s", string(debug.Stack()))
+			}
+		}()
+		if s.isStop {
+			return
+		}
+		timer := time.NewTicker(time.Second)
+		for {
+			<-timer.C
+			gcKeys := []string{}
+			s.udpConns.IterCb(func(key string, v interface{}) {
+				if time.Now().Unix()-v.(*UDPConnItem).touchtime > gctime {
+					(*(v.(*UDPConnItem).conn)).Close()
+					(v.(*UDPConnItem).udpConn).Close()
+					gcKeys = append(gcKeys, key)
+					s.log.Printf("gc udp conn %s", key)
+				}
+			})
+			for _, k := range gcKeys {
+				s.udpConns.Remove(k)
+			}
+			gcKeys = nil
+		}
+	}()
 }
 func (s *TCP) GetParentConn() (conn net.Conn, err error) {
 	if *s.cfg.ParentType == "tls" {

services/udp/udp.go

@@ -1,10 +1,8 @@
 package udp
 
 import (
-	"bufio"
 	"crypto/tls"
 	"fmt"
-	"hash/crc32"
 	"io"
 	logger "log"
 	"net"
@@ -35,12 +33,30 @@ type UDP struct {
 	sc               *utils.ServerChannel
 	isStop           bool
 	log              *logger.Logger
+	outUDPConnCtxMap mapx.ConcurrentMap
+	udpConns         mapx.ConcurrentMap
+	dstAddr          *net.UDPAddr
+}
+type UDPConnItem struct {
+	conn      *net.Conn
+	touchtime int64
+	srcAddr   *net.UDPAddr
+	localAddr *net.UDPAddr
+	connid    string
+}
+type outUDPConnCtx struct {
+	localAddr *net.UDPAddr
+	srcAddr   *net.UDPAddr
+	udpconn   *net.UDPConn
+	touchtime int64
 }
 
 func NewUDP() services.Service {
 	return &UDP{
 		p:                mapx.NewConcurrentMap(),
 		isStop:           false,
+		outUDPConnCtxMap: mapx.NewConcurrentMap(),
+		udpConns:         mapx.NewConcurrentMap(),
 	}
 }
 func (s *UDP) CheckArgs() (err error) {
@@ -49,7 +65,7 @@ func (s *UDP) CheckArgs() (err error) {
 		return
 	}
 	if *s.cfg.ParentType == "" {
-		err = fmt.Errorf("parent type unkown,use -T <tls|tcp>")
+		err = fmt.Errorf("parent type unkown,use -T <udp|tls|tcp>")
 		return
 	}
 	if *s.cfg.ParentType == "tls" {
@@ -58,10 +74,17 @@ func (s *UDP) CheckArgs() (err error) {
 			return
 		}
 	}
+
+	s.dstAddr, err = net.ResolveUDPAddr("udp", *s.cfg.Parent)
+	if err != nil {
+		s.log.Printf("resolve udp addr %s fail  fail,ERR:%s", *s.cfg.Parent, err)
+		return
+	}
 	return
 }
 func (s *UDP) InitService() (err error) {
-
+	s.OutToUDPGCDeamon()
+	s.UDPGCDeamon()
 	return
 }
 func (s *UDP) StopService() {
@@ -117,19 +140,15 @@ func (s *UDP) callback(listener *net.UDPConn, packet []byte, localAddr, srcAddr
 			s.log.Printf("udp conn handler crashed with err : %s \nstack: %s", err, string(debug.Stack()))
 		}
 	}()
-	var err error
 	switch *s.cfg.ParentType {
 	case "tcp":
 		fallthrough
 	case "tls":
-		err = s.OutToTCP(packet, localAddr, srcAddr)
+		s.OutToTCP(packet, localAddr, srcAddr)
 	case "udp":
-		err = s.OutToUDP(packet, localAddr, srcAddr)
+		s.OutToUDP(packet, localAddr, srcAddr)
 	default:
-		err = fmt.Errorf("unkown parent type %s", *s.cfg.ParentType)
+		s.log.Printf("unkown parent type %s", *s.cfg.ParentType)
-	}
-	if err != nil {
-		s.log.Printf("connect to %s parent %s fail, ERR:%s", *s.cfg.ParentType, *s.cfg.Parent, err)
 	}
 }
 func (s *UDP) GetConn(connKey string) (conn net.Conn, isNew bool, err error) {
@@ -147,104 +166,97 @@ func (s *UDP) GetConn(connKey string) (conn net.Conn, isNew bool, err error) {
 	conn = _conn.(net.Conn)
 	return
 }
-func (s *UDP) OutToTCP(packet []byte, localAddr, srcAddr *net.UDPAddr) (err error) {
+func (s *UDP) OutToTCP(data []byte, localAddr, srcAddr *net.UDPAddr) (err error) {
-	numLocal := crc32.ChecksumIEEE([]byte(localAddr.String()))
+	s.UDPSend(data, localAddr, srcAddr)
-	numSrc := crc32.ChecksumIEEE([]byte(srcAddr.String()))
-	mod := uint32(10)
-	if mod == 0 {
-		mod = 10
-	}
-	connKey := uint64((numLocal/10)*10 + numSrc%mod)
-	conn, isNew, err := s.GetConn(fmt.Sprintf("%d", connKey))
-	if err != nil {
-		s.log.Printf("upd get conn to %s parent %s fail, ERR:%s", *s.cfg.ParentType, *s.cfg.Parent, err)
-		return
-	}
-	if isNew {
-		go func() {
-			defer func() {
-				if err := recover(); err != nil {
-					s.log.Printf("udp conn handler out to tcp crashed with err : %s \nstack: %s", err, string(debug.Stack()))
-				}
-			}()
-			s.log.Printf("conn %d created , local: %s", connKey, srcAddr.String())
-			for {
-				if s.isStop {
-					conn.Close()
-					return
-				}
-				srcAddrFromConn, body, err := utils.ReadUDPPacket(bufio.NewReader(conn))
-				if err == io.EOF || err == io.ErrUnexpectedEOF {
-					//s.log.Printf("connection %d released", connKey)
-					s.p.Remove(fmt.Sprintf("%d", connKey))
-					break
-				}
-				if err != nil {
-					s.log.Printf("parse revecived udp packet fail, err: %s", err)
-					continue
-				}
-				//s.log.Printf("udp packet revecived over parent , local:%s", srcAddrFromConn)
-				_srcAddr := strings.Split(srcAddrFromConn, ":")
-				if len(_srcAddr) != 2 {
-					s.log.Printf("parse revecived udp packet fail, addr error : %s", srcAddrFromConn)
-					continue
-				}
-				port, _ := strconv.Atoi(_srcAddr[1])
-				dstAddr := &net.UDPAddr{IP: net.ParseIP(_srcAddr[0]), Port: port}
-				_, err = s.sc.UDPListener.WriteToUDP(body, dstAddr)
-				if err != nil {
-					s.log.Printf("udp response to local %s fail,ERR:%s", srcAddr, err)
-					continue
-				}
-				//s.log.Printf("udp response to local %s success", srcAddr)
-			}
-		}()
-	}
-	//s.log.Printf("select conn %d , local: %s", connKey, srcAddr.String())
-	writer := bufio.NewWriter(conn)
-	//fmt.Println(conn, writer)
-	writer.Write(utils.UDPPacket(srcAddr.String(), packet))
-	err = writer.Flush()
-	if err != nil {
-		s.log.Printf("write udp packet to %s fail ,flush err:%s", *s.cfg.Parent, err)
-		return
-	}
-	//s.log.Printf("write packet %v", packet)
 	return
 }
-func (s *UDP) OutToUDP(packet []byte, localAddr, srcAddr *net.UDPAddr) (err error) {
+func (s *UDP) OutToUDPGCDeamon() {
-	//s.log.Printf("udp packet revecived:%s,%v", srcAddr, packet)
+	gctime := int64(30)
-	dstAddr, err := net.ResolveUDPAddr("udp", *s.cfg.Parent)
+	go func() {
-	if err != nil {
+		defer func() {
-		s.log.Printf("resolve udp addr %s fail  fail,ERR:%s", dstAddr.String(), err)
+			if e := recover(); e != nil {
+				fmt.Printf("crashed:%s", string(debug.Stack()))
+			}
+		}()
+		if s.isStop {
 			return
 		}
+		timer := time.NewTicker(time.Second)
+		for {
+			<-timer.C
+			gcKeys := []string{}
+			s.outUDPConnCtxMap.IterCb(func(key string, v interface{}) {
+				if time.Now().Unix()-v.(*outUDPConnCtx).touchtime > gctime {
+					(*(v.(*outUDPConnCtx).udpconn)).Close()
+					gcKeys = append(gcKeys, key)
+					s.log.Printf("gc udp conn %s <--> %s", (*v.(*outUDPConnCtx)).srcAddr, (*v.(*outUDPConnCtx)).localAddr)
+				}
+			})
+			for _, k := range gcKeys {
+				s.outUDPConnCtxMap.Remove(k)
+			}
+			gcKeys = nil
+		}
+	}()
+}
+func (s *UDP) OutToUDP(packet []byte, localAddr, srcAddr *net.UDPAddr) {
+	var ouc *outUDPConnCtx
+	if v, ok := s.outUDPConnCtxMap.Get(srcAddr.String()); !ok {
 		clientSrcAddr := &net.UDPAddr{IP: net.IPv4zero, Port: 0}
-	conn, err := net.DialUDP("udp", clientSrcAddr, dstAddr)
+		conn, err := net.DialUDP("udp", clientSrcAddr, s.dstAddr)
 		if err != nil {
-		s.log.Printf("connect to udp %s fail,ERR:%s", dstAddr.String(), err)
+			s.log.Printf("connect to udp %s fail,ERR:%s", s.dstAddr.String(), err)
+
+		}
+		ouc = &outUDPConnCtx{
+			localAddr: localAddr,
+			srcAddr:   srcAddr,
+			udpconn:   conn,
+		}
+		s.outUDPConnCtxMap.Set(srcAddr.String(), ouc)
+		go func() {
+			defer func() {
+				if e := recover(); e != nil {
+					fmt.Printf("crashed:%s", string(debug.Stack()))
+				}
+			}()
+			s.log.Printf("udp conn %s <--> %s connected", srcAddr.String(), localAddr.String())
+			buf := utils.LeakyBuffer.Get()
+			defer func() {
+				utils.LeakyBuffer.Put(buf)
+				s.outUDPConnCtxMap.Remove(srcAddr.String())
+				s.log.Printf("udp conn %s <--> %s released", srcAddr.String(), localAddr.String())
+			}()
+			for {
+				n, err := ouc.udpconn.Read(buf)
+				if err != nil {
+					if !utils.IsNetClosedErr(err) {
+						s.log.Printf("udp conn read udp packet fail , err: %s ", err)
+					}
 					return
 				}
-	conn.SetDeadline(time.Now().Add(time.Millisecond * time.Duration(*s.cfg.Timeout)))
+				ouc.touchtime = time.Now().Unix()
-	_, err = conn.Write(packet)
+				go func() {
-	if err != nil {
+					defer func() {
-		s.log.Printf("send udp packet to %s fail,ERR:%s", dstAddr.String(), err)
+						if e := recover(); e != nil {
-		return
+							fmt.Printf("crashed:%s", string(debug.Stack()))
 						}
-	//s.log.Printf("send udp packet to %s success", dstAddr.String())
+					}()
-	buf := make([]byte, 512)
+					(*(s.sc).UDPListener).SetWriteDeadline(time.Now().Add(time.Millisecond * time.Duration(*s.cfg.Timeout)))
-	len, _, err := conn.ReadFromUDP(buf)
+					_, err = (*(s.sc).UDPListener).WriteTo(buf[:n], srcAddr)
-	if err != nil {
+					(*(s.sc).UDPListener).SetWriteDeadline(time.Time{})
-		s.log.Printf("read udp response from %s fail ,ERR:%s", dstAddr.String(), err)
+				}()
-		return
 			}
-	//s.log.Printf("revecived udp packet from %s , %v", dstAddr.String(), respBody)
+		}()
-	_, err = s.sc.UDPListener.WriteToUDP(buf[0:len], srcAddr)
+	} else {
-	if err != nil {
+		ouc = v.(*outUDPConnCtx)
-		s.log.Printf("send udp response to cluster fail ,ERR:%s", err)
-		return
 	}
-	//s.log.Printf("send udp response to cluster success ,from:%s", dstAddr.String())
+	go func() {
+		ouc.touchtime = time.Now().Unix()
+		ouc.udpconn.SetWriteDeadline(time.Now().Add(time.Millisecond * time.Duration(*s.cfg.Timeout)))
+		ouc.udpconn.Write(packet)
+		ouc.udpconn.SetWriteDeadline(time.Time{})
+	}()
 	return
 }
 func (s *UDP) GetParentConn() (conn net.Conn, err error) {
@@ -259,3 +271,126 @@ func (s *UDP) GetParentConn() (conn net.Conn, err error) {
 	}
 	return
 }
+func (s *UDP) UDPGCDeamon() {
+	gctime := int64(30)
+	go func() {
+		defer func() {
+			if e := recover(); e != nil {
+				fmt.Printf("crashed:%s", string(debug.Stack()))
+			}
+		}()
+		if s.isStop {
+			return
+		}
+		timer := time.NewTicker(time.Second)
+		for {
+			<-timer.C
+			gcKeys := []string{}
+			s.udpConns.IterCb(func(key string, v interface{}) {
+				if time.Now().Unix()-v.(*UDPConnItem).touchtime > gctime {
+					(*(v.(*UDPConnItem).conn)).Close()
+					gcKeys = append(gcKeys, key)
+					s.log.Printf("gc udp conn %s", v.(*UDPConnItem).connid)
+				}
+			})
+			for _, k := range gcKeys {
+				s.udpConns.Remove(k)
+			}
+			gcKeys = nil
+		}
+	}()
+}
+func (s *UDP) UDPSend(data []byte, localAddr, srcAddr *net.UDPAddr) {
+	var (
+		uc      *UDPConnItem
+		key     = srcAddr.String()
+		err     error
+		outconn net.Conn
+	)
+	v, ok := s.udpConns.Get(key)
+	if !ok {
+		for {
+			outconn, err = s.GetParentConn()
+			if err != nil && strings.Contains(err.Error(), "can not connect at same time") {
+				time.Sleep(time.Millisecond * 500)
+				continue
+			} else {
+				break
+			}
+		}
+		if err != nil {
+			s.log.Printf("connect to %s fail, err: %s", *s.cfg.Parent, err)
+			return
+		}
+		uc = &UDPConnItem{
+			conn:      &outconn,
+			srcAddr:   srcAddr,
+			localAddr: localAddr,
+		}
+		s.udpConns.Set(key, uc)
+		s.UDPRevecive(key)
+	} else {
+		uc = v.(*UDPConnItem)
+	}
+	go func() {
+		defer func() {
+			if e := recover(); e != nil {
+				(*uc.conn).Close()
+				s.udpConns.Remove(key)
+				s.log.Printf("udp sender crashed with error : %s", e)
+			}
+		}()
+		uc.touchtime = time.Now().Unix()
+		(*uc.conn).SetWriteDeadline(time.Now().Add(time.Millisecond * time.Duration(*s.cfg.Timeout)))
+		_, err = (*uc.conn).Write(utils.UDPPacket(fmt.Sprintf("%s", srcAddr.String()), data))
+		(*uc.conn).SetWriteDeadline(time.Time{})
+		if err != nil {
+			s.log.Printf("write udp packet to %s fail ,flush err:%s ", *s.cfg.Parent, err)
+		}
+	}()
+}
+func (s *UDP) UDPRevecive(key string) {
+	go func() {
+		defer func() {
+			if e := recover(); e != nil {
+				fmt.Printf("crashed:%s", string(debug.Stack()))
+			}
+		}()
+		s.log.Printf("udp conn %s connected", key)
+		var uc *UDPConnItem
+		defer func() {
+			if uc != nil {
+				(*uc.conn).Close()
+			}
+			s.udpConns.Remove(key)
+			s.log.Printf("udp conn %s released", key)
+		}()
+		v, ok := s.udpConns.Get(key)
+		if !ok {
+			s.log.Printf("[warn] udp conn not exists for %s, connid : %s", key)
+			return
+		}
+		uc = v.(*UDPConnItem)
+		for {
+			_, body, err := utils.ReadUDPPacket(*uc.conn)
+			if err != nil {
+				if strings.Contains(err.Error(), "n != int(") {
+					continue
+				}
+				if err != io.EOF && !utils.IsNetClosedErr(err) {
+					s.log.Printf("udp conn read udp packet fail , err: %s ", err)
+				}
+				return
+			}
+			uc.touchtime = time.Now().Unix()
+			go func() {
+				defer func() {
+					if e := recover(); e != nil {
+						fmt.Printf("crashed:%s", string(debug.Stack()))
+					}
+				}()
+				s.sc.UDPListener.WriteToUDP(body, uc.srcAddr)
+			}()
+		}
+	}()
+}

utils/mapx/map.go

@@ -192,13 +192,12 @@ func snapshot(m ConcurrentMap) (chans []chan Tuple) {
 	wg.Add(SHARD_COUNT)
 	// Foreach shard.
 	for index, shard := range m {
-		go func() {
+		go func(index int, shard *ConcurrentMapShared) {
 			defer func() {
 				if e := recover(); e != nil {
 					fmt.Printf("crashed:%s", string(debug.Stack()))
 				}
 			}()
-			func(index int, shard *ConcurrentMapShared) {
 			// Foreach key, value pair.
 			shard.RLock()
 			chans[index] = make(chan Tuple, len(shard.items))
@@ -209,7 +208,6 @@ func snapshot(m ConcurrentMap) (chans []chan Tuple) {
 			shard.RUnlock()
 			close(chans[index])
 		}(index, shard)
-		}()
 	}
 	wg.Wait()
 	return chans