go-gost/x
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add buffer size option for udp connection

Browse Source
This commit is contained in:
ginuerzh
2025-07-30 21:40:53 +08:00
parent b64e5901b6
commit a5309eee97
26 changed files with 218 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
handler/http/metadata.go
+2
View File
@@ -25,6 +25,7 @@ type metadata struct {
compression bool
probeResistance *probeResistance
enableUDP bool
udpBufferSize int
header http.Header
hash string
authBasicRealm string
@@ -77,6 +78,7 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error {
}
}
h.md.enableUDP = mdutil.GetBool(md, "udp")
h.md.udpBufferSize = mdutil.GetInt(md, "udp.bufferSize", "udpBufferSize")
h.md.hash = mdutil.GetString(md, "hash")
h.md.authBasicRealm = mdutil.GetString(md, "authBasicRealm")
handler/http/udp.go
+1
View File
@@ -75,6 +75,7 @@ func (h *httpHandler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecorde
relay := udp.NewRelay(socks.UDPTunServerConn(conn), pc).
WithBypass(h.options.Bypass).
WithBufferSize(h.md.udpBufferSize).
WithLogger(log)
t := time.Now()
handler/relay/bind.go
+1
View File
@@ -205,6 +205,7 @@ func (h *relayHandler) bindUDP(ctx context.Context, conn net.Conn, network, addr
r := udp.NewRelay(relay_util.UDPTunServerConn(conn), pc).
WithBypass(h.options.Bypass).
WithBufferSize(h.md.udpBufferSize).
WithLogger(log)
t := time.Now()
handler/relay/metadata.go
+7 -5
View File
@@ -14,11 +14,12 @@ import (
)
type metadata struct {
readTimeout time.Duration
enableBind bool
noDelay bool
hash string
muxCfg *mux.Config
readTimeout time.Duration
udpBufferSize int
enableBind bool
noDelay bool
hash string
muxCfg *mux.Config
observerPeriod time.Duration
observerResetTraffic bool
@@ -43,6 +44,7 @@ func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) {
h.md.readTimeout = 15 * time.Second
}
h.md.udpBufferSize = mdutil.GetInt(md, "udp.bufferSize", "udpBufferSize")
h.md.enableBind = mdutil.GetBool(md, "bind")
h.md.noDelay = mdutil.GetBool(md, "nodelay")
h.md.hash = mdutil.GetString(md, "hash")
handler/router/associate.go
+4 -2
View File
@@ -121,9 +121,11 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw
log.Debugf("%s/%s: router=%s, connector=%s, weight=%d established", host, network, routerID, connectorID, connectorID.Weight())
var b [MaxMessageSize]byte
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
for {
n, err := conn.Read(b[:])
n, err := conn.Read(b)
if err != nil {
if err == io.EOF {
return nil
handler/router/entrypoint.go
+4 -2
View File
@@ -3,16 +3,18 @@ package router
import (
"bytes"
"github.com/go-gost/core/common/bufpool"
"github.com/go-gost/core/logger"
"github.com/go-gost/relay"
)
func (h *routerHandler) handleEntrypoint(log logger.Logger) error {
var buf [MaxMessageSize]byte
buf := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(buf)
for {
err := func() error {
n, addr, err := h.epConn.ReadFrom(buf[:])
n, addr, err := h.epConn.ReadFrom(buf)
if err != nil {
return err
}
handler/router/metadata.go
+6 -2
View File
@@ -1,7 +1,6 @@
package router
import (
"math"
"time"
"github.com/go-gost/core/ingress"
@@ -13,13 +12,14 @@ import (
)
const (
MaxMessageSize = math.MaxUint16
defaultTTL = 15 * time.Second
defaultBufferSize = 4096
defaultCacheExpiration = time.Second
)
type metadata struct {
readTimeout time.Duration
bufferSize int
entryPoint string
ingress ingress.Ingress
@@ -40,6 +40,10 @@ type metadata struct {
func (h *routerHandler) parseMetadata(md mdata.Metadata) (err error) {
h.md.readTimeout = mdutil.GetDuration(md, "readTimeout")
h.md.bufferSize = mdutil.GetInt(md, "router.bufferSize", "bufferSize")
if h.md.bufferSize <= 0 {
h.md.bufferSize = defaultBufferSize
}
h.md.entryPoint = mdutil.GetString(md, "entrypoint")
h.md.ingress = registry.IngressRegistry().Get(mdutil.GetString(md, "ingress"))
handler/socks/v5/handler.go
+2 -1
View File
@@ -145,6 +145,7 @@ func (h *socks5Handler) Handle(ctx context.Context, conn net.Conn, opts ...handl
}
log.WithFields(map[string]any{
"network": ro.Network,
"duration": time.Since(start),
"inputBytes": ro.InputBytes,
"outputBytes": ro.OutputBytes,
@@ -186,7 +187,7 @@ func (h *socks5Handler) Handle(ctx context.Context, conn net.Conn, opts ...handl
return h.handleMuxBind(ctx, conn, "tcp", address, ro, log)
case gosocks5.CmdUdp:
ro.Network = "udp"
return h.handleUDP(ctx, conn, ro, log)
return h.handleUDP(ctx, conn, "udp", ro, log)
case socks.CmdUDPTun:
ro.Network = "udp"
return h.handleUDPTun(ctx, conn, "udp", address, ro, log)
handler/socks/v5/metadata.go
+2
View File
@@ -19,6 +19,7 @@ type metadata struct {
noTLS bool
enableBind bool
enableUDP bool
udpBufferSize int
compatibilityMode bool
hash string
muxCfg *mux.Config
@@ -50,6 +51,7 @@ func (h *socks5Handler) parseMetadata(md mdata.Metadata) (err error) {
h.md.noTLS = mdutil.GetBool(md, "notls")
h.md.enableBind = mdutil.GetBool(md, "bind")
h.md.enableUDP = mdutil.GetBool(md, "udp")
h.md.udpBufferSize = mdutil.GetInt(md, "udp.bufferSize", "udpBufferSize")
h.md.compatibilityMode = mdutil.GetBool(md, "comp")
h.md.hash = mdutil.GetString(md, "hash")
handler/socks/v5/udp.go
+8 -6
View File
@@ -23,9 +23,10 @@ import (
xrecorder "github.com/go-gost/x/recorder"
)
func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, network string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
log = log.WithFields(map[string]any{
"cmd": "udp",
"network": network,
"cmd": network,
})
if !h.md.enableUDP {
@@ -39,7 +40,7 @@ func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecor
Netns: h.options.Netns,
}
cc, err := lc.ListenPacket(ctx, "udp", "")
cc, err := lc.ListenPacket(ctx, network, "")
if err != nil {
log.Error(err)
reply := gosocks5.NewReply(gosocks5.Failure, nil)
@@ -74,7 +75,7 @@ func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecor
// obtain a udp connection
var buf bytes.Buffer
c, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "udp", "") // UDP association
c, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), network, "") // UDP association
ro.Route = buf.String()
if err != nil {
log.Error(err)
@@ -106,7 +107,7 @@ func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecor
string(clientID),
limiter.ServiceOption(h.options.Service),
limiter.ScopeOption(limiter.ScopeClient),
limiter.NetworkOption("udp"),
limiter.NetworkOption(network),
limiter.ClientOption(string(clientID)),
limiter.SrcOption(conn.RemoteAddr().String()),
)
@@ -119,8 +120,9 @@ func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, ro *xrecor
}
}
r := udp.NewRelay(socks.UDPConn(cc), pc).
r := udp.NewRelay(socks.UDPConn(cc, h.md.udpBufferSize), pc).
WithBypass(h.options.Bypass).
WithBufferSize(h.md.udpBufferSize).
WithLogger(log)
go r.Run(ctx)
handler/socks/v5/udp_tun.go
+5 -3
View File
@@ -22,7 +22,8 @@ import (
func (h *socks5Handler) handleUDPTun(ctx context.Context, conn net.Conn, network, address string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
log = log.WithFields(map[string]any{
"cmd": "udp-tun",
"network": network,
"cmd": "udp-tun",
})
{
@@ -66,7 +67,7 @@ func (h *socks5Handler) handleUDPTun(ctx context.Context, conn net.Conn, network
// obtain a udp connection
var buf bytes.Buffer
c, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "udp", "") // UDP association
c, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), network, "") // UDP association
ro.Route = buf.String()
if err != nil {
log.Error(err)
@@ -94,7 +95,7 @@ func (h *socks5Handler) handleUDPTun(ctx context.Context, conn net.Conn, network
Netns: h.options.Netns,
}
var err error
pc, err = lc.ListenPacket(ctx, "udp", bindAddr.String())
pc, err = lc.ListenPacket(ctx, network, bindAddr.String())
if err != nil {
log.Error(err)
reply := gosocks5.NewReply(gosocks5.Failure, nil)
@@ -133,6 +134,7 @@ func (h *socks5Handler) handleUDPTun(ctx context.Context, conn net.Conn, network
r := udp.NewRelay(socks.UDPTunServerConn(conn), pc).
WithBypass(h.options.Bypass).
WithBufferSize(h.md.udpBufferSize).
WithLogger(log)
t := time.Now()
handler/ss/udp/handler.go
+15 -5
View File
@@ -7,6 +7,7 @@ import (
"net"
"time"
"github.com/go-gost/core/common/bufpool"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
@@ -112,7 +113,7 @@ func (h *ssuHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
pc = h.cipher.PacketConn(pc)
}
// standard UDP relay.
pc = ss.UDPServerConn(pc, conn.RemoteAddr())
pc = ss.UDPServerConn(pc, conn.RemoteAddr(), h.md.udpBufferSize)
} else {
if h.cipher != nil {
conn = ss.ShadowConn(h.cipher.StreamConn(conn), nil)
@@ -150,11 +151,18 @@ func (h *ssuHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
func (h *ssuHandler) relayPacket(ctx context.Context, pc1, pc2 net.PacketConn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) {
errc := make(chan error, 2)
bufferSize := h.md.udpBufferSize
if bufferSize <= 0 {
bufferSize = defaultBufferSize
}
go func() {
var b [MaxMessageSize]byte
b := bufpool.Get(bufferSize)
defer bufpool.Put(b)
for {
err := func() error {
n, addr, err := pc1.ReadFrom(b[:])
n, addr, err := pc1.ReadFrom(b)
if err != nil {
return err
}
@@ -185,10 +193,12 @@ func (h *ssuHandler) relayPacket(ctx context.Context, pc1, pc2 net.PacketConn, r
}()
go func() {
var b [MaxMessageSize]byte
b := bufpool.Get(bufferSize)
defer bufpool.Put(b)
for {
err := func() error {
n, raddr, err := pc2.ReadFrom(b[:])
n, raddr, err := pc2.ReadFrom(b)
if err != nil {
return err
}
handler/ss/udp/metadata.go
+7 -10
View File
@@ -1,7 +1,6 @@
package ss
import (
"math"
"time"
mdata "github.com/go-gost/core/metadata"
@@ -9,22 +8,20 @@ import (
)
const (
MaxMessageSize = math.MaxUint16
defaultBufferSize = 4096
)
type metadata struct {
key string
readTimeout time.Duration
key string
readTimeout time.Duration
udpBufferSize int
}
func (h *ssuHandler) parseMetadata(md mdata.Metadata) (err error) {
const (
key = "key"
readTimeout = "readTimeout"
)
h.md.key = mdutil.GetString(md, key)
h.md.readTimeout = mdutil.GetDuration(md, readTimeout)
h.md.key = mdutil.GetString(md, "key")
h.md.readTimeout = mdutil.GetDuration(md, "readTimeout")
h.md.udpBufferSize = mdutil.GetInt(md, "udpBufferSize", "udp.bufferSize")
return
}
handler/tungo/metadata.go
+7 -1
View File
@@ -10,8 +10,13 @@ import (
mdutil "github.com/go-gost/x/metadata/util"
)
const (
defaultBufferSize = 4096
)
type metadata struct {
udpTimeout time.Duration
udpTimeout time.Duration
udpBufferSize int
sniffing bool
sniffingUDP bool
@@ -34,6 +39,7 @@ type metadata struct {
func (h *tungoHandler) parseMetadata(md mdata.Metadata) (err error) {
h.md.udpTimeout = mdutil.GetDuration(md, "udpTimeout", "tungo.udpTimeout")
h.md.udpBufferSize = mdutil.GetInt(md, "udp.bufferSize", "udpBufferSize")
h.md.sniffing = mdutil.GetBool(md, "sniffing")
h.md.sniffingUDP = mdutil.GetBool(md, "sniffing.udp")
handler/tungo/stack.go
+22 -8
View File
@@ -6,13 +6,13 @@ import (
"context"
"crypto/tls"
"io"
"math"
"net"
"net/netip"
"strings"
"sync"
"time"
"github.com/go-gost/core/common/bufpool"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/observer/stats"
"github.com/go-gost/core/recorder"
@@ -44,7 +44,8 @@ type transportHandler struct {
procCancel context.CancelFunc
// UDP session timeout.
udpTimeout time.Duration
udpTimeout time.Duration
udpBufferSize int
sniffing bool
sniffingUDP bool
@@ -344,33 +345,46 @@ func (h *transportHandler) handleUDPConn(uc adapter.UDPConn) {
t := time.Now()
log.Infof("%s <-> %s", remoteAddr, dstAddr)
pipePacketData(conn, cc, h.sniffingUDP, ro, h.udpTimeout)
h.pipePacketData(conn, cc, ro)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Infof("%s >-< %s", remoteAddr, dstAddr)
}
func pipePacketData(conn1, conn2 net.Conn, sniffing bool, ro *xrecorder.HandlerRecorderObject, timeout time.Duration) {
func (h *transportHandler) pipePacketData(conn1, conn2 net.Conn, ro *xrecorder.HandlerRecorderObject) {
timeout := h.udpTimeout
if timeout <= 0 {
timeout = udpSessionTimeout
}
bufferSize := h.udpBufferSize
if bufferSize <= 0 {
bufferSize = defaultBufferSize
}
wg := sync.WaitGroup{}
wg.Add(2)
go func() {
defer wg.Done()
copyPacketData(conn1, conn2, sniffing, false, ro, timeout)
buf := bufpool.Get(bufferSize)
defer bufpool.Put(buf)
copyPacketData(conn1, conn2, buf, h.sniffing, false, ro, timeout)
}()
go func() {
defer wg.Done()
copyPacketData(conn2, conn1, sniffing, true, ro, timeout)
buf := bufpool.Get(bufferSize)
defer bufpool.Put(buf)
copyPacketData(conn2, conn1, buf, h.sniffing, true, ro, timeout)
}()
wg.Wait()
}
func copyPacketData(dst, src net.Conn, sniffing bool, c2s bool, ro *xrecorder.HandlerRecorderObject, timeout time.Duration) error {
buf := make([]byte, math.MaxUint16/2)
func copyPacketData(dst, src net.Conn, buf []byte, sniffing bool, c2s bool, ro *xrecorder.HandlerRecorderObject, timeout time.Duration) error {
isDNS := false
for {