ptr

common/singbridge/destination.go

@@ -3,7 +3,6 @@ package singbridge
 import (
 	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
-	"github.com/xtls/xray-core/common/errors"
 	"github.com/xtls/xray-core/common/net"
 )
 
@@ -18,14 +17,14 @@ func ToNetwork(network string) net.Network {
 	}
 }
 
-func ToDestination(socksaddr M.Socksaddr, network net.Network) (net.Destination, error) {
+func ToDestination(socksaddr M.Socksaddr, network net.Network) net.Destination {
 	// IsFqdn() implicitly checks if the domain name is valid
 	if socksaddr.IsFqdn() {
 		return net.Destination{
 			Network: network,
 			Address: net.DomainAddress(socksaddr.Fqdn),
 			Port:    net.Port(socksaddr.Port),
-		}, nil
+		}
 	}
 
 	// IsIP() implicitly checks if the IP address is valid
@@ -34,10 +33,10 @@ func ToDestination(socksaddr M.Socksaddr, network net.Network) (net.Destination,
 			Network: network,
 			Address: net.IPAddress(socksaddr.Addr.AsSlice()),
 			Port:    net.Port(socksaddr.Port),
-		}, nil
+		}
 	}
 
-	return net.Destination{}, errors.New("invalid socks address: ", socksaddr)
+	return net.Destination{}
 }
 
 func ToSocksaddr(destination net.Destination) M.Socksaddr {

common/singbridge/dialer.go

@@ -26,11 +26,7 @@ func NewDialer(dialer internet.Dialer) *XrayDialer {
 }
 
 func (d *XrayDialer) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
-	dest, err := ToDestination(destination, ToNetwork(network))
-	if err != nil {
-		return nil, err
-	}
-	return d.Dialer.Dial(ctx, dest)
+	return d.Dialer.Dial(ctx, ToDestination(destination, ToNetwork(network)))
 }
 
 func (d *XrayDialer) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) {
@@ -47,17 +43,13 @@ func NewOutboundDialer(outbound proxy.Outbound, dialer internet.Dialer) *XrayOut
 }
 
 func (d *XrayOutboundDialer) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
-	dest, err := ToDestination(destination, ToNetwork(network))
-	if err != nil {
-		return nil, err
-	}
 	outbounds := session.OutboundsFromContext(ctx)
 	if len(outbounds) == 0 {
 		outbounds = []*session.Outbound{{}}
 		ctx = session.ContextWithOutbounds(ctx, outbounds)
 	}
 	ob := outbounds[len(outbounds)-1]
-	ob.Target = dest
+	ob.Target = ToDestination(destination, ToNetwork(network))
 
 	opts := []pipe.Option{pipe.WithSizeLimit(64 * 1024)}
 	uplinkReader, uplinkWriter := pipe.New(opts...)

common/singbridge/handler.go

@@ -31,23 +31,15 @@ func NewDispatcher(dispatcher routing.Dispatcher, newErrorFunc func(values ...an
 }
 
 func (d *Dispatcher) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
-	dest, err := ToDestination(metadata.Destination, net.Network_TCP)
-	if err != nil {
-		return err
-	}
 	xConn := NewConn(conn)
-	return d.upstream.DispatchLink(ctx, dest, &transport.Link{
+	return d.upstream.DispatchLink(ctx, ToDestination(metadata.Destination, net.Network_TCP), &transport.Link{
 		Reader: xConn,
 		Writer: xConn,
 	})
 }
 
 func (d *Dispatcher) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error {
-	dest, err := ToDestination(metadata.Destination, net.Network_UDP)
-	if err != nil {
-		return err
-	}
-	return d.upstream.DispatchLink(ctx, dest, &transport.Link{
+	return d.upstream.DispatchLink(ctx, ToDestination(metadata.Destination, net.Network_UDP), &transport.Link{
 		Reader: buf.NewPacketReader(conn.(io.Reader)),
 		Writer: buf.NewWriter(conn.(io.Writer)),
 	})

common/singbridge/packet.go

@@ -10,21 +10,15 @@ import (
 	"github.com/xtls/xray-core/common"
 	"github.com/xtls/xray-core/common/buf"
 	"github.com/xtls/xray-core/common/net"
-	"github.com/xtls/xray-core/common/signal"
 	"github.com/xtls/xray-core/transport"
 )
 
 func CopyPacketConn(ctx context.Context, inboundConn net.Conn, link *transport.Link, destination net.Destination, serverConn net.PacketConn) error {
-	cancel := func() {
-		common.Interrupt(link.Reader)
-		common.Interrupt(serverConn)
-	}
 	conn := &PacketConnWrapper{
 		Reader: link.Reader,
 		Writer: link.Writer,
 		Dest:   destination,
 		Conn:   inboundConn,
-		T:      signal.CancelAfterInactivity(ctx, cancel, 300*time.Second),
 	}
 	return ReturnError(bufio.CopyPacketConn(ctx, conn, bufio.NewPacketConn(serverConn)))
 }
@@ -35,19 +29,11 @@ type PacketConnWrapper struct {
 	net.Conn
 	Dest   net.Destination
 	cached buf.MultiBuffer
-
-	// A simple patch to avoid goroutine leak since sing infra cannot awake read block by write err
-	T *signal.ActivityTimer
 }
 
-func (w *PacketConnWrapper) ReadPacket(buffer *B.Buffer) (addr M.Socksaddr, err error) {
-	w.T.Update()
-	defer func() {
-		if err != nil {
-			// uplinkonly
-			w.T.SetTimeout(2 * time.Second)
-		}
-	}()
+// This ReadPacket implemented a timeout to avoid goroutine leak like PipeConnWrapper.Read()
+// as a temporarily solution
+func (w *PacketConnWrapper) ReadPacket(buffer *B.Buffer) (M.Socksaddr, error) {
 	if w.cached != nil {
 		mb, bb := buf.SplitFirst(w.cached)
 		if bb == nil {
@@ -65,7 +51,30 @@ func (w *PacketConnWrapper) ReadPacket(buffer *B.Buffer) (addr M.Socksaddr, err
 			return ToSocksaddr(destination), nil
 		}
 	}
-	mb, err := w.ReadMultiBuffer()
+
+	// timeout
+	type readResult struct {
+		mb  buf.MultiBuffer
+		err error
+	}
+	c := make(chan readResult, 1)
+	go func() {
+		mb, err := w.ReadMultiBuffer()
+		c <- readResult{mb: mb, err: err}
+	}()
+	var mb buf.MultiBuffer
+	select {
+	case <-time.After(60 * time.Second):
+		common.Close(w.Reader)
+		common.Interrupt(w.Reader)
+		return M.Socksaddr{}, buf.ErrReadTimeout
+	case result := <-c:
+		if result.err != nil {
+			return M.Socksaddr{}, result.err
+		}
+		mb = result.mb
+	}
+
 	nb, bb := buf.SplitFirst(mb)
 	if bb == nil {
 		return M.Socksaddr{}, nil
@@ -83,22 +92,12 @@ func (w *PacketConnWrapper) ReadPacket(buffer *B.Buffer) (addr M.Socksaddr, err
 	}
 }
 
-func (w *PacketConnWrapper) WritePacket(buffer *B.Buffer, destination M.Socksaddr) (err error) {
-	w.T.Update()
-	defer func() {
-		if err != nil {
-			// downlinkonly
-			w.T.SetTimeout(5 * time.Second)
-		}
-	}()
-	endpoint, err := ToDestination(destination, net.Network_UDP)
-	if err != nil {
-		return err
-	}
+func (w *PacketConnWrapper) WritePacket(buffer *B.Buffer, destination M.Socksaddr) error {
 	vBuf := buf.New()
 	vBuf.Write(buffer.Bytes())
+	endpoint := ToDestination(destination, net.Network_UDP)
 	vBuf.UDP = &endpoint
-	return w.WriteMultiBuffer(buf.MultiBuffer{vBuf})
+	return w.Writer.WriteMultiBuffer(buf.MultiBuffer{vBuf})
 }
 
 func (w *PacketConnWrapper) Close() error {

common/singbridge/pipe.go

@@ -9,7 +9,6 @@ import (
 	"github.com/sagernet/sing/common/bufio"
 	"github.com/xtls/xray-core/common"
 	"github.com/xtls/xray-core/common/buf"
-	"github.com/xtls/xray-core/common/signal"
 	"github.com/xtls/xray-core/transport"
 )
 
@@ -23,11 +22,6 @@ func CopyConn(ctx context.Context, inboundConn net.Conn, link *transport.Link, s
 	} else {
 		conn.R = &buf.BufferedReader{Reader: link.Reader}
 	}
-	cancel := func() {
-		common.Interrupt(link.Reader)
-		common.Interrupt(serverConn)
-	}
-	conn.T = signal.CancelAfterInactivity(ctx, cancel, 300*time.Second)
 	return ReturnError(bufio.CopyConn(ctx, conn, serverConn))
 }
 
@@ -35,27 +29,35 @@ type PipeConnWrapper struct {
 	R io.Reader
 	W buf.Writer
 	net.Conn
-
-	// A simple patch to avoid goroutine leak since sing infra cannot awake read block by write err
-	T *signal.ActivityTimer
 }
 
 func (w *PipeConnWrapper) Close() error {
 	return nil
 }
 
+// This Read implemented a timeout to avoid goroutine leak.
+// as a temporarily solution
 func (w *PipeConnWrapper) Read(b []byte) (n int, err error) {
-	w.T.Update()
-	n, err = w.R.Read(b)
-	if err != nil {
-		// uplinkonly
-		w.T.SetTimeout(2 * time.Second)
+	type readResult struct {
+		n   int
+		err error
+	}
+	c := make(chan readResult, 1)
+	go func() {
+		n, err := w.R.Read(b)
+		c <- readResult{n: n, err: err}
+	}()
+	select {
+	case result := <-c:
+		return result.n, result.err
+	case <-time.After(300 * time.Second):
+		common.Close(w.R)
+		common.Interrupt(w.R)
+		return 0, buf.ErrReadTimeout
 	}
-	return
 }
 
 func (w *PipeConnWrapper) Write(p []byte) (n int, err error) {
-	w.T.Update()
 	n = len(p)
 	var mb buf.MultiBuffer
 	pLen := len(p)
@@ -74,8 +76,6 @@ func (w *PipeConnWrapper) Write(p []byte) (n int, err error) {
 	if err != nil {
 		n = 0
 		buf.ReleaseMulti(mb)
-		// downlinkonly
-		w.T.SetTimeout(5 * time.Second)
 	}
 	return
 }

proxy/shadowsocks_2022/inbound.go

@@ -2,7 +2,6 @@ package shadowsocks_2022
 
 import (
 	"context"
-	"time"
 
 	shadowsocks "github.com/sagernet/sing-shadowsocks"
 	"github.com/sagernet/sing-shadowsocks/shadowaead_2022"
@@ -19,7 +18,6 @@ import (
 	"github.com/xtls/xray-core/common/net"
 	"github.com/xtls/xray-core/common/protocol"
 	"github.com/xtls/xray-core/common/session"
-	"github.com/xtls/xray-core/common/signal"
 	"github.com/xtls/xray-core/common/singbridge"
 	"github.com/xtls/xray-core/features/routing"
 	"github.com/xtls/xray-core/transport/internet/stat"
@@ -117,11 +115,7 @@ func (i *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata M.M
 	})
 	errors.LogInfo(ctx, "tunnelling request to tcp:", metadata.Destination)
 	dispatcher := session.DispatcherFromContext(ctx)
-	destination, err := singbridge.ToDestination(metadata.Destination, net.Network_TCP)
-	if err != nil {
-		return err
-	}
-	link, err := dispatcher.Dispatch(ctx, destination)
+	link, err := dispatcher.Dispatch(ctx, singbridge.ToDestination(metadata.Destination, net.Network_TCP))
 	if err != nil {
 		return err
 	}
@@ -142,10 +136,7 @@ func (i *Inbound) NewPacketConnection(ctx context.Context, conn N.PacketConn, me
 	})
 	errors.LogInfo(ctx, "tunnelling request to udp:", metadata.Destination)
 	dispatcher := session.DispatcherFromContext(ctx)
-	destination, err := singbridge.ToDestination(metadata.Destination, net.Network_UDP)
-	if err != nil {
-		return err
-	}
+	destination := singbridge.ToDestination(metadata.Destination, net.Network_UDP)
 	link, err := dispatcher.Dispatch(ctx, destination)
 	if err != nil {
 		return err
@@ -154,9 +145,6 @@ func (i *Inbound) NewPacketConnection(ctx context.Context, conn N.PacketConn, me
 		Reader: link.Reader,
 		Writer: link.Writer,
 		Dest:   destination,
-		T: signal.CancelAfterInactivity(ctx, func() {
-			common.Interrupt(link.Reader)
-		}, 300*time.Second),
 	}
 	return bufio.CopyPacketConn(ctx, conn, outConn)
 }

proxy/shadowsocks_2022/inbound_multi.go

@@ -6,7 +6,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/sagernet/sing-shadowsocks/shadowaead_2022"
 	C "github.com/sagernet/sing/common"
@@ -23,7 +22,6 @@ import (
 	"github.com/xtls/xray-core/common/net"
 	"github.com/xtls/xray-core/common/protocol"
 	"github.com/xtls/xray-core/common/session"
-	"github.com/xtls/xray-core/common/signal"
 	"github.com/xtls/xray-core/common/singbridge"
 	"github.com/xtls/xray-core/common/uuid"
 	"github.com/xtls/xray-core/features/routing"
@@ -239,10 +237,11 @@ func (i *MultiUserInbound) NewConnection(ctx context.Context, conn net.Conn, met
 	})
 	errors.LogInfo(ctx, "tunnelling request to tcp:", metadata.Destination)
 	dispatcher := session.DispatcherFromContext(ctx)
-	destination, err := singbridge.ToDestination(metadata.Destination, net.Network_TCP)
-	if err != nil {
-		return err
+	destination := singbridge.ToDestination(metadata.Destination, net.Network_TCP)
+	if !destination.IsValid() {
+		return errors.New("invalid destination")
 	}
+
 	link, err := dispatcher.Dispatch(ctx, destination)
 	if err != nil {
 		return err
@@ -263,10 +262,7 @@ func (i *MultiUserInbound) NewPacketConnection(ctx context.Context, conn N.Packe
 	})
 	errors.LogInfo(ctx, "tunnelling request to udp:", metadata.Destination)
 	dispatcher := session.DispatcherFromContext(ctx)
-	destination, err := singbridge.ToDestination(metadata.Destination, net.Network_UDP)
-	if err != nil {
-		return err
-	}
+	destination := singbridge.ToDestination(metadata.Destination, net.Network_UDP)
 	link, err := dispatcher.Dispatch(ctx, destination)
 	if err != nil {
 		return err
@@ -275,9 +271,6 @@ func (i *MultiUserInbound) NewPacketConnection(ctx context.Context, conn N.Packe
 		Reader: link.Reader,
 		Writer: link.Writer,
 		Dest:   destination,
-		T: signal.CancelAfterInactivity(ctx, func() {
-			common.Interrupt(link.Reader)
-		}, 300*time.Second),
 	}
 	return bufio.CopyPacketConn(ctx, conn, outConn)
 }

proxy/shadowsocks_2022/inbound_relay.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/sagernet/sing-shadowsocks/shadowaead_2022"
 	C "github.com/sagernet/sing/common"
@@ -21,7 +20,6 @@ import (
 	"github.com/xtls/xray-core/common/net"
 	"github.com/xtls/xray-core/common/protocol"
 	"github.com/xtls/xray-core/common/session"
-	"github.com/xtls/xray-core/common/signal"
 	"github.com/xtls/xray-core/common/singbridge"
 	"github.com/xtls/xray-core/common/uuid"
 	"github.com/xtls/xray-core/features/routing"
@@ -140,11 +138,7 @@ func (i *RelayInbound) NewConnection(ctx context.Context, conn net.Conn, metadat
 	})
 	errors.LogInfo(ctx, "tunnelling request to tcp:", metadata.Destination)
 	dispatcher := session.DispatcherFromContext(ctx)
-	destination, err := singbridge.ToDestination(metadata.Destination, net.Network_TCP)
-	if err != nil {
-		return err
-	}
-	link, err := dispatcher.Dispatch(ctx, destination)
+	link, err := dispatcher.Dispatch(ctx, singbridge.ToDestination(metadata.Destination, net.Network_TCP))
 	if err != nil {
 		return err
 	}
@@ -167,10 +161,7 @@ func (i *RelayInbound) NewPacketConnection(ctx context.Context, conn N.PacketCon
 	})
 	errors.LogInfo(ctx, "tunnelling request to udp:", metadata.Destination)
 	dispatcher := session.DispatcherFromContext(ctx)
-	destination, err := singbridge.ToDestination(metadata.Destination, net.Network_UDP)
-	if err != nil {
-		return err
-	}
+	destination := singbridge.ToDestination(metadata.Destination, net.Network_UDP)
 	link, err := dispatcher.Dispatch(ctx, destination)
 	if err != nil {
 		return err
@@ -179,9 +170,6 @@ func (i *RelayInbound) NewPacketConnection(ctx context.Context, conn N.PacketCon
 		Reader: link.Reader,
 		Writer: link.Writer,
 		Dest:   destination,
-		T: signal.CancelAfterInactivity(ctx, func() {
-			common.Interrupt(link.Reader)
-		}, 300*time.Second),
 	}
 	return bufio.CopyPacketConn(ctx, conn, outConn)
 }

proxy/shadowsocks_2022/outbound.go

@@ -16,7 +16,6 @@ import (
 	"github.com/xtls/xray-core/common/errors"
 	"github.com/xtls/xray-core/common/net"
 	"github.com/xtls/xray-core/common/session"
-	"github.com/xtls/xray-core/common/signal"
 	"github.com/xtls/xray-core/common/singbridge"
 	"github.com/xtls/xray-core/transport"
 	"github.com/xtls/xray-core/transport/internet"
@@ -143,9 +142,6 @@ func (o *Outbound) Process(ctx context.Context, link *transport.Link, dialer int
 				Writer: link.Writer,
 				Conn:   inboundConn,
 				Dest:   destination,
-				T: signal.CancelAfterInactivity(ctx, func() {
-					common.Interrupt(link.Reader)
-				}, 300*time.Second),
 			}
 		}
 

transport/internet/splithttp/dialer.go

@@ -183,6 +183,8 @@ func createHTTPClient(dest net.Destination, streamSettings *internet.MemoryStrea
 		if quicParams.KeepAlivePeriod == 0 {
 			if keepAlivePeriod == 0 {
 				quicConfig.KeepAlivePeriod = net.QuicgoH3KeepAlivePeriod
+			} else if keepAlivePeriod > 0 {
+				quicConfig.KeepAlivePeriod = keepAlivePeriod
 			}
 		}
 		if quicParams.MaxIncomingStreams == 0 {
@@ -506,6 +508,8 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me
 		var seq int64
 		var lastWrite time.Time
 
+		dynamicHTTPClient := httpClient
+		dynamicXmuxClient := xmuxClient
 		for {
 			// by offloading the uploads into a buffered pipe, multiple conn.Write
 			// calls get automatically batched together into larger POST requests.
@@ -540,13 +544,13 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me
 
 				lastWrite = time.Now()
 
-				if xmuxClient != nil && (xmuxClient.LeftRequests.Add(-1) <= 0 ||
-					(xmuxClient.UnreusableAt != time.Time{} && lastWrite.After(xmuxClient.UnreusableAt))) {
-					httpClient, xmuxClient = getHTTPClient(ctx, dest, streamSettings)
+				if dynamicXmuxClient != nil && (dynamicXmuxClient.LeftRequests.Add(-1) <= 0 ||
+					(dynamicXmuxClient.UnreusableAt != time.Time{} && lastWrite.After(dynamicXmuxClient.UnreusableAt))) {
+					dynamicHTTPClient, dynamicXmuxClient = getHTTPClient(ctx, dest, streamSettings)
 				}
 
-				go func() {
-					err := httpClient.PostPacket(
+				go func(hClient DialerClient) {
+					err := hClient.PostPacket(
 						ctx,
 						requestURL.String(),
 						sessionId,
@@ -559,9 +563,9 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me
 						uploadPipeReader.Interrupt()
 						doSplit.Store(false)
 					}
-				}()
+				}(dynamicHTTPClient)
 
-				if _, ok := httpClient.(*DefaultDialerClient); ok {
+				if _, ok := dynamicHTTPClient.(*DefaultDialerClient); ok {
 					<-wroteRequest.Wait()
 				}
 			}