fix udp connection timeout

This commit is contained in:
ginuerzh 2024-08-06 18:33:29 +08:00
parent 5e8a8a4b4d
commit 22e522e933
9 changed files with 30 additions and 24 deletions

View File

@ -78,7 +78,7 @@ func (*defaultRoute) Bind(ctx context.Context, network, address string, opts ...
ReadQueueSize: options.UDPDataQueueSize, ReadQueueSize: options.UDPDataQueueSize,
ReadBufferSize: options.UDPDataBufferSize, ReadBufferSize: options.UDPDataBufferSize,
TTL: options.UDPConnTTL, TTL: options.UDPConnTTL,
KeepAlive: true, Keepalive: true,
Logger: logger, Logger: logger,
}) })
return ln, err return ln, err

View File

@ -73,7 +73,7 @@ func (c *relayConnector) bindUDP(ctx context.Context, conn net.Conn, network, ad
ReadQueueSize: opts.UDPDataQueueSize, ReadQueueSize: opts.UDPDataQueueSize,
ReadBufferSize: opts.UDPDataBufferSize, ReadBufferSize: opts.UDPDataBufferSize,
TTL: opts.UDPConnTTL, TTL: opts.UDPConnTTL,
KeepAlive: true, Keepalive: true,
Logger: log, Logger: log,
}) })

View File

@ -87,7 +87,7 @@ func (c *socks5Connector) bindUDP(ctx context.Context, conn net.Conn, network, a
ReadQueueSize: opts.UDPDataQueueSize, ReadQueueSize: opts.UDPDataQueueSize,
ReadBufferSize: opts.UDPDataBufferSize, ReadBufferSize: opts.UDPDataBufferSize,
TTL: opts.UDPConnTTL, TTL: opts.UDPConnTTL,
KeepAlive: true, Keepalive: true,
Logger: log, Logger: log,
}) })

View File

@ -83,6 +83,9 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error {
log.Trace(string(dump)) log.Trace(string(dump))
} }
resp.ProtoMajor = req.ProtoMajor
resp.ProtoMinor = req.ProtoMinor
var tunnelID relay.TunnelID var tunnelID relay.TunnelID
if ep.ingress != nil { if ep.ingress != nil {
if rule := ep.ingress.GetRule(ctx, req.Host); rule != nil { if rule := ep.ingress.GetRule(ctx, req.Host); rule != nil {

View File

@ -17,17 +17,16 @@ type ListenConfig struct {
ReadQueueSize int ReadQueueSize int
ReadBufferSize int ReadBufferSize int
TTL time.Duration TTL time.Duration
KeepAlive bool Keepalive bool
Logger logger.Logger Logger logger.Logger
} }
type listener struct { type listener struct {
conn net.PacketConn conn net.PacketConn
cqueue chan net.Conn cqueue chan net.Conn
connPool *connPool connPool *connPool
// mux sync.Mutex closed chan struct{}
closed chan struct{} errChan chan error
errChan chan error config *ListenConfig
config *ListenConfig
} }
func NewListener(conn net.PacketConn, cfg *ListenConfig) net.Listener { func NewListener(conn net.PacketConn, cfg *ListenConfig) net.Listener {
@ -42,9 +41,7 @@ func NewListener(conn net.PacketConn, cfg *ListenConfig) net.Listener {
errChan: make(chan error, 1), errChan: make(chan error, 1),
config: cfg, config: cfg,
} }
if cfg.KeepAlive { ln.connPool = newConnPool(cfg.TTL).WithLogger(cfg.Logger)
ln.connPool = newConnPool(cfg.TTL).WithLogger(cfg.Logger)
}
go ln.listenLoop() go ln.listenLoop()
return ln return ln
@ -113,15 +110,12 @@ func (ln *listener) Close() error {
} }
func (ln *listener) getConn(raddr net.Addr) *conn { func (ln *listener) getConn(raddr net.Addr) *conn {
// ln.mux.Lock()
// defer ln.mux.Unlock()
c, ok := ln.connPool.Get(raddr.String()) c, ok := ln.connPool.Get(raddr.String())
if ok { if ok && !c.isClosed() {
return c return c
} }
c = newConn(ln.conn, ln.Addr(), raddr, ln.config.ReadQueueSize, ln.config.KeepAlive) c = newConn(ln.conn, ln.Addr(), raddr, ln.config.ReadQueueSize, ln.config.Keepalive)
select { select {
case ln.cqueue <- c: case ln.cqueue <- c:
ln.connPool.Set(raddr.String(), c) ln.connPool.Set(raddr.String(), c)
@ -142,17 +136,17 @@ type conn struct {
idle int32 // indicate the connection is idle idle int32 // indicate the connection is idle
closed chan struct{} closed chan struct{}
closeMutex sync.Mutex closeMutex sync.Mutex
keepAlive bool keepalive bool
} }
func newConn(c net.PacketConn, laddr, remoteAddr net.Addr, queueSize int, keepAlive bool) *conn { func newConn(c net.PacketConn, laddr, remoteAddr net.Addr, queueSize int, keepalive bool) *conn {
return &conn{ return &conn{
PacketConn: c, PacketConn: c,
localAddr: laddr, localAddr: laddr,
remoteAddr: remoteAddr, remoteAddr: remoteAddr,
rc: make(chan []byte, queueSize), rc: make(chan []byte, queueSize),
closed: make(chan struct{}), closed: make(chan struct{}),
keepAlive: keepAlive, keepalive: keepalive,
} }
} }
@ -179,7 +173,7 @@ func (c *conn) Read(b []byte) (n int, err error) {
} }
func (c *conn) WriteTo(b []byte, addr net.Addr) (n int, err error) { func (c *conn) WriteTo(b []byte, addr net.Addr) (n int, err error) {
if !c.keepAlive { if !c.keepalive {
defer c.Close() defer c.Close()
} }
return c.PacketConn.WriteTo(b, addr) return c.PacketConn.WriteTo(b, addr)
@ -201,6 +195,15 @@ func (c *conn) Close() error {
return nil return nil
} }
func (c *conn) isClosed() bool {
select {
case <-c.closed:
return true
default:
return false
}
}
func (c *conn) LocalAddr() net.Addr { func (c *conn) LocalAddr() net.Addr {
return c.localAddr return c.localAddr
} }

View File

@ -64,7 +64,7 @@ func (l *ftcpListener) Init(md md.Metadata) (err error) {
ReadQueueSize: l.md.readQueueSize, ReadQueueSize: l.md.readQueueSize,
ReadBufferSize: l.md.readBufferSize, ReadBufferSize: l.md.readBufferSize,
TTL: l.md.ttl, TTL: l.md.ttl,
KeepAlive: true, Keepalive: true,
Logger: l.logger, Logger: l.logger,
}) })
return return

View File

@ -9,7 +9,7 @@ import (
const ( const (
defaultTTL = 5 * time.Second defaultTTL = 5 * time.Second
defaultReadBufferSize = 1024 defaultReadBufferSize = 8192
defaultReadQueueSize = 1024 defaultReadQueueSize = 1024
defaultBacklog = 128 defaultBacklog = 128
) )

View File

@ -65,7 +65,7 @@ func (l *udpListener) Init(md md.Metadata) (err error) {
Backlog: l.md.backlog, Backlog: l.md.backlog,
ReadQueueSize: l.md.readQueueSize, ReadQueueSize: l.md.readQueueSize,
ReadBufferSize: l.md.readBufferSize, ReadBufferSize: l.md.readBufferSize,
KeepAlive: l.md.keepalive, Keepalive: l.md.keepalive,
TTL: l.md.ttl, TTL: l.md.ttl,
Logger: l.logger, Logger: l.logger,
}) })

View File

@ -9,7 +9,7 @@ import (
const ( const (
defaultTTL = 5 * time.Second defaultTTL = 5 * time.Second
defaultReadBufferSize = 1024 defaultReadBufferSize = 8192
defaultReadQueueSize = 128 defaultReadQueueSize = 128
defaultBacklog = 128 defaultBacklog = 128
) )