Compare commits

..
7 Commits
Author SHA1 Message Date
null 2b0d898d08 use RLock 2026-10-08 23:38:14 +08:00
null 8f705a1ea4 fix logic plus 2026-10-08 22:35:35 +08:00
null 9897b5a9d3 refine logic 2026-10-08 21:12:51 +08:00
null 6bcf81b0ab fix logic 2026-10-08 20:38:06 +08:00
null 542030ca71 chore 2026-10-08 19:30:46 +08:00
Fangliding 0d83d099ac make client unavailable if instance stopped 2026-10-08 19:03:48 +08:00
Fangliding d06cf79243 Hysteria: Remove closed instance conn 2026-10-08 15:35:39 +08:00
3 changed files with 47 additions and 55 deletions
+3 -1
View File
@@ -92,6 +92,8 @@ type Instance struct {
// Instance state
func (server *Instance) IsRunning() bool {
server.statusLock.Lock()
defer server.statusLock.Unlock()
return server.running
}
@@ -320,7 +322,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.IsRunning() {
if err := feature.Start(); err != nil {
errors.LogInfoInner(s.ctx, err, "failed to start feature")
}
+2 -10
View File
@@ -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()
}
+42 -44
View File
@@ -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"
@@ -29,6 +29,8 @@ import (
type client struct {
sync.Mutex
instance *core.Instance
forced bool
dest net.Destination
config *Config
tlsConfig *gotls.Config
@@ -36,12 +38,10 @@ 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
}
func (c *client) status() status {
@@ -67,11 +67,14 @@ func (c *client) close() {
}
func (c *client) dial(ctx context.Context) error {
status := c.status()
if status == StatusActive {
return nil
if c.forced {
return errors.New("client is closed")
}
if status == StatusInactive {
switch c.status() {
case StatusActive:
return nil
case StatusInactive:
c.close()
}
@@ -93,7 +96,7 @@ func (c *client) dial(ctx context.Context) error {
ChromeParrot: !quicParams.DisableChromeParrot,
EnableDatagrams: true,
MaxDatagramFrameSize: MaxDatagramFrameSize,
OmitMaxDatagramFrameSize: time.Now().After(time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)),
OmitMaxDatagramFrameSize: true,
DisablePathManager: true,
}
if quicParams.InitStreamReceiveWindow == 0 {
@@ -238,15 +241,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 +262,15 @@ func (c *client) udp(ctx context.Context) (stat.Connection, error) {
return c.udpSM.udp()
}
func (c *client) clean() {
func (c *client) clean(force bool) {
c.Lock()
defer c.Unlock()
switch c.status() {
case StatusInactive:
if force {
c.forced = true
}
if status := c.status(); force && status != StatusNull || status == StatusInactive {
c.close()
return
case StatusNull:
return
}
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()
}
c.Unlock()
}
type dialerConf struct {
@@ -301,11 +286,23 @@ type clientManager struct {
func (m *clientManager) clean() {
ticker := time.NewTicker(idleCleanupInterval)
for range ticker.C {
var forced []dialerConf
m.RLock()
for _, c := range m.m {
c.clean()
for k, c := range m.m {
force := c.instance != nil && !c.instance.IsRunning()
c.clean(force)
if force {
forced = append(forced, k)
}
}
m.RUnlock()
for i := range forced {
m.Lock()
delete(m.m, forced[i])
m.Unlock()
}
}
}
@@ -330,25 +327,26 @@ 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()
}