fix: xray内核删除用户时重复加锁导致卡死

core/xray/app/dispatcher/default.go

@@ -102,13 +102,13 @@ func (r *cachedReader) Interrupt() {
 
 // DefaultDispatcher is a default implementation of Dispatcher.
 type DefaultDispatcher struct {
-	ohm     outbound.Manager
-	router  routing.Router
-	policy  policy.Manager
-	stats   stats.Manager
-	fdns    dns.FakeDNSEngine
-	Wm      *WriterManager
-	Counter sync.Map
+	ohm          outbound.Manager
+	router       routing.Router
+	policy       policy.Manager
+	stats        stats.Manager
+	fdns         dns.FakeDNSEngine
+	Counter      sync.Map
+	LinkManagers sync.Map // map[string]*LinkManager
 }
 
 func init() {
@@ -132,9 +132,6 @@ func (d *DefaultDispatcher) Init(config *Config, om outbound.Manager, router rou
 	d.router = router
 	d.policy = pm
 	d.stats = sm
-	d.Wm = &WriterManager{
-		writers: make(map[string]map[*ManagedWriter]struct{}),
-	}
 	return nil
 }
 
@@ -197,14 +194,23 @@ func (d *DefaultDispatcher) getLink(ctx context.Context, network net.Network) (*
 			common.Interrupt(inboundLink.Reader)
 			return nil, nil, nil, errors.New("Limited ", user.Email, " by conn or ip")
 		}
+		var lm *LinkManager
+		if lmloaded, ok := d.LinkManagers.Load(user.Email); !ok {
+			lm = &LinkManager{
+				links: make(map[*ManagedWriter]buf.Reader),
+			}
+			d.LinkManagers.Store(user.Email, lm)
+		} else {
+			lm = lmloaded.(*LinkManager)
+		}
 		managedWriter := &ManagedWriter{
 			writer:  uplinkWriter,
-			email:   user.Email,
-			manager: d.Wm,
+			manager: lm,
 		}
-		d.Wm.AddWriter(managedWriter)
+		lm.AddLink(managedWriter, outboundLink.Reader)
 		inboundLink.Writer = managedWriter
 		if w != nil {
+			sessionInbound.CanSpliceCopy = 3
 			inboundLink.Writer = rate.NewRateLimitWriter(inboundLink.Writer, w)
 			outboundLink.Writer = rate.NewRateLimitWriter(outboundLink.Writer, w)
 		}
@@ -382,14 +388,22 @@ func (d *DefaultDispatcher) DispatchLink(ctx context.Context, destination net.De
 			common.Interrupt(outbound.Reader)
 			return errors.New("Limited ", user.Email, " by conn or ip")
 		}
+		var lm *LinkManager
+		if lmloaded, ok := d.LinkManagers.Load(user.Email); !ok {
+			lm = &LinkManager{
+				links: make(map[*ManagedWriter]buf.Reader),
+			}
+			d.LinkManagers.Store(user.Email, lm)
+		} else {
+			lm = lmloaded.(*LinkManager)
+		}
 		managedWriter := &ManagedWriter{
 			writer:  outbound.Writer,
-			email:   user.Email,
-			manager: d.Wm,
+			manager: lm,
 		}
-		d.Wm.AddWriter(managedWriter)
 		outbound.Writer = managedWriter
 		if w != nil {
+			sessionInbound.CanSpliceCopy = 3
 			outbound.Writer = rate.NewRateLimitWriter(outbound.Writer, w)
 		}
 		var t *counter.TrafficCounter
@@ -406,6 +420,7 @@ func (d *DefaultDispatcher) DispatchLink(ctx context.Context, destination net.De
 			Reader:  &buf.TimeoutWrapperReader{Reader: outbound.Reader},
 			Counter: &ts.UpCounter,
 		}
+		lm.AddLink(managedWriter, outbound.Reader)
 		outbound.Writer = &dispatcher.SizeStatWriter{
 			Counter: downcounter,
 			Writer:  outbound.Writer,

core/xray/app/dispatcher/linkmanager.go

@@ -7,48 +7,9 @@ import (
 	"github.com/xtls/xray-core/common/buf"
 )
 
-type WriterManager struct {
-	writers map[string]map[*ManagedWriter]struct{}
-	mu      sync.Mutex
-}
-
-func (m *WriterManager) AddWriter(writer *ManagedWriter) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if _, exists := m.writers[writer.email]; !exists {
-		m.writers[writer.email] = make(map[*ManagedWriter]struct{})
-	}
-	m.writers[writer.email][writer] = struct{}{}
-}
-
-func (m *WriterManager) RemoveWriter(writer *ManagedWriter) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if _, exists := m.writers[writer.email]; !exists {
-		return
-	}
-	delete(m.writers[writer.email], writer)
-}
-
-func (m *WriterManager) RemoveWritersForUser(email string) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if _, exists := m.writers[email]; !exists {
-		return
-	}
-	for writer := range m.writers[email] {
-		delete(m.writers[email], writer)
-		common.Close(writer.writer)
-	}
-	delete(m.writers, email)
-}
-
 type ManagedWriter struct {
 	writer  buf.Writer
-	email   string
-	manager *WriterManager
+	manager *LinkManager
 }
 
 func (w *ManagedWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
@@ -59,3 +20,27 @@ func (w *ManagedWriter) Close() error {
 	w.manager.RemoveWriter(w)
 	return common.Close(w.writer)
 }
+
+type LinkManager struct {
+	links map[*ManagedWriter]buf.Reader
+	mu    sync.Mutex
+}
+
+func (m *LinkManager) AddLink(writer *ManagedWriter, reader buf.Reader) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.links[writer] = reader
+}
+
+func (m *LinkManager) RemoveWriter(writer *ManagedWriter) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.links, writer)
+}
+
+func (m *LinkManager) CloseAll() {
+	for w, r := range m.links {
+		common.Close(w)
+		common.Interrupt(r)
+	}
+}

core/xray/user.go

@@ -8,6 +8,7 @@ import (
 	"github.com/InazumaV/V2bX/common/counter"
 	"github.com/InazumaV/V2bX/common/format"
 	vCore "github.com/InazumaV/V2bX/core"
+	"github.com/InazumaV/V2bX/core/xray/app/dispatcher"
 	"github.com/xtls/xray-core/common/protocol"
 	"github.com/xtls/xray-core/proxy"
 )
@@ -43,8 +44,15 @@ func (c *Xray) DelUsers(users []panel.UserInfo, tag string, _ *panel.NodeInfo) e
 			return err
 		}
 		delete(c.users.uidMap, user)
-		c.dispatcher.Counter.Delete(user)
-		c.dispatcher.Wm.RemoveWritersForUser(user)
+		if v, ok := c.dispatcher.Counter.Load(tag); ok {
+			tc := v.(*counter.TrafficCounter)
+			tc.Delete(user)
+		}
+		if v, ok := c.dispatcher.LinkManagers.Load(user); ok {
+			lm := v.(*dispatcher.LinkManager)
+			lm.CloseAll()
+			c.dispatcher.LinkManagers.Delete(user)
+		}
 	}
 	return nil
 }