Hysteria: Remove closed instance conn

core/xray.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"reflect"
 	"sync"
+	"sync/atomic"
 
 	"github.com/xtls/xray-core/common"
 	"github.com/xtls/xray-core/common/errors"
@@ -84,7 +85,7 @@ type Instance struct {
 	features                   []features.Feature
 	pendingResolutions         []resolution
 	pendingOptionalResolutions []resolution
-	running                    bool
+	running                    atomic.Bool
 	resolveLock                sync.Mutex
 
 	ctx context.Context
@@ -92,7 +93,7 @@ type Instance struct {
 
 // Instance state
 func (server *Instance) IsRunning() bool {
-	return server.running
+	return server.running.Load()
 }
 
 func AddInboundHandler(server *Instance, config *InboundHandlerConfig) error {
@@ -262,7 +263,7 @@ func (s *Instance) Close() error {
 	s.statusLock.Lock()
 	defer s.statusLock.Unlock()
 
-	s.running = false
+	s.running.Store(false)
 
 	var errs []interface{}
 	for _, f := range s.features {
@@ -320,7 +321,7 @@ func (s *Instance) RequireFeatures(callback interface{}, optional bool) error {
 
 // AddFeature registers a feature into current Instance.
 func (s *Instance) AddFeature(feature features.Feature) error {
-	if s.running {
+	if s.running.Load() {
 		if err := feature.Start(); err != nil {
 			errors.LogInfoInner(s.ctx, err, "failed to start feature")
 		}
@@ -388,7 +389,7 @@ func (s *Instance) Start() error {
 	s.statusLock.Lock()
 	defer s.statusLock.Unlock()
 
-	s.running = true
+	s.running.Store(true)
 	for _, f := range s.features {
 		if err := f.Start(); err != nil {
 			return err

transport/internet/hysteria/conn.go

@@ -5,7 +5,6 @@ import (
 	"encoding/binary"
 	"io"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/apernet/quic-go"
@@ -21,10 +20,8 @@ type interConn struct {
 	local  net.Addr
 	remote net.Addr
 
-	client    bool
-	user      *protocol.MemoryUser
-	closeOnce sync.Once
-	aliveTCP  *atomic.Int64
+	client bool
+	user   *protocol.MemoryUser
 }
 
 func (c *interConn) User() *protocol.MemoryUser {
@@ -49,11 +46,6 @@ func (c *interConn) Write(b []byte) (int, error) {
 
 func (c *interConn) Close() error {
 	c.stream.CancelRead(0)
-	if c.aliveTCP != nil {
-		c.closeOnce.Do(func() {
-			c.aliveTCP.Add(-1)
-		})
-	}
 	return c.stream.Close()
 }
 

transport/internet/hysteria/dialer.go

@@ -9,7 +9,6 @@ import (
 	"runtime"
 	"strconv"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/apernet/quic-go"
@@ -18,6 +17,7 @@ import (
 	"github.com/xtls/xray-core/common/errors"
 	"github.com/xtls/xray-core/common/net"
 	"github.com/xtls/xray-core/common/net/cnc"
+	"github.com/xtls/xray-core/core"
 	"github.com/xtls/xray-core/transport/internet"
 	"github.com/xtls/xray-core/transport/internet/finalmask"
 	"github.com/xtls/xray-core/transport/internet/hysteria/congestion"
@@ -36,12 +36,11 @@ type client struct {
 	finalMask    *finalmask.FinalMask
 	quicParams   *internet.QuicParams
 
-	conn      *quic.Conn
-	tr        *quic.Transport
-	pktConn   net.PacketConn
-	udpSM     *udpSessionManager
-	aliveTCP  *atomic.Int64
-	lastAlive time.Time
+	conn     *quic.Conn
+	tr       *quic.Transport
+	pktConn  net.PacketConn
+	udpSM    *udpSessionManager
+	instance *core.Instance
 }
 
 func (c *client) status() status {
@@ -57,9 +56,11 @@ func (c *client) status() status {
 }
 
 func (c *client) close() {
-	c.conn.CloseWithError(closeErrCodeOK, "")
-	c.tr.Close()
-	c.pktConn.Close()
+	if c.conn != nil {
+		c.conn.CloseWithError(closeErrCodeOK, "")
+	}
+	common.CloseIfExists(c.tr)
+	common.CloseIfExists(c.pktConn)
 	c.conn = nil
 	c.tr = nil
 	c.pktConn = nil
@@ -238,15 +239,12 @@ func (c *client) tcp(ctx context.Context) (stat.Connection, error) {
 		return nil, err
 	}
 
-	c.lastAlive = time.Now()
-	c.aliveTCP.Add(1)
 	return &interConn{
 		stream: stream,
 		local:  c.conn.LocalAddr(),
 		remote: c.conn.RemoteAddr(),
 
-		client:   true,
-		aliveTCP: c.aliveTCP,
+		client: true,
 	}, nil
 }
 
@@ -262,30 +260,18 @@ func (c *client) udp(ctx context.Context) (stat.Connection, error) {
 	return c.udpSM.udp()
 }
 
-func (c *client) clean() {
+func (c *client) clean() (shouldDelete bool) {
 	c.Lock()
 	defer c.Unlock()
-	switch c.status() {
-	case StatusInactive:
+	if c.instance != nil && !c.instance.IsRunning() {
 		c.close()
-		return
-	case StatusNull:
-		return
+		return true
 	}
-	var udpSessions int
-	if c.udpSM != nil {
-		c.udpSM.RLock()
-		udpSessions = len(c.udpSM.m)
-		c.udpSM.RUnlock()
-	}
-	if udpSessions == 0 && c.aliveTCP.Load() == 0 {
-		if c.lastAlive.Add(net.ConnIdleTimeout).Before(time.Now()) {
-			c.close()
-			return
-		}
-	} else {
-		c.lastAlive = time.Now()
+	if c.status() == StatusInactive {
+		c.close()
+		return false
 	}
+	return false
 }
 
 type dialerConf struct {
@@ -301,11 +287,21 @@ type clientManager struct {
 func (m *clientManager) clean() {
 	ticker := time.NewTicker(idleCleanupInterval)
 	for range ticker.C {
+		var toDelete []dialerConf
 		m.RLock()
-		for _, c := range m.m {
-			c.clean()
+		for k, c := range m.m {
+			if c.clean() {
+				toDelete = append(toDelete, k)
+			}
 		}
 		m.RUnlock()
+		if len(toDelete) > 0 {
+			m.Lock()
+			for _, k := range toDelete {
+				delete(m.m, k)
+			}
+			m.Unlock()
+		}
 	}
 }
 
@@ -330,25 +326,25 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me
 		go manager.clean()
 	})
 
+	dialerConfKey := dialerConf{dest, streamSettings}
 	manager.RLock()
-	c := manager.m[dialerConf{dest, streamSettings}]
+	c := manager.m[dialerConfKey]
 	manager.RUnlock()
 
 	if c == nil {
 		manager.Lock()
-		c = manager.m[dialerConf{dest, streamSettings}]
+		c = manager.m[dialerConfKey]
 		if c == nil {
 			c = &client{
+				instance:     core.FromContext(ctx),
 				dest:         dest,
 				config:       streamSettings.ProtocolSettings.(*Config),
 				tlsConfig:    tlsConfig.GetTLSConfig(tls.WithDestination(dest)),
 				socketConfig: streamSettings.SocketSettings,
 				finalMask:    streamSettings.FinalMask,
 				quicParams:   streamSettings.QuicParams,
-				aliveTCP:     &atomic.Int64{},
-				lastAlive:    time.Now(),
 			}
-			manager.m[dialerConf{dest, streamSettings}] = c
+			manager.m[dialerConfKey] = c
 		}
 		manager.Unlock()
 	}