世界
f0f2617b79
The URL test history update hook and the Clash mode update hook were single-slot: the API service's attached service overwrote the hook set by the daemon, so clients stopped receiving group updates. Replace both with multicast hook lists. Also share a single URL test history storage via context: Clash API looked it up under a key nobody registered and fell back to its own empty storage, so dashboards showed no delay once an API service was configured. Selector changes now notify through the shared storage, covering selections made from any API surface.
428 lines
13 KiB
Go
428 lines
13 KiB
Go
package group
|
|
|
|
import (
|
|
"context"
|
|
"net"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/sagernet/sing-box/adapter"
|
|
"github.com/sagernet/sing-box/adapter/outbound"
|
|
"github.com/sagernet/sing-box/common/interrupt"
|
|
"github.com/sagernet/sing-box/common/urltest"
|
|
C "github.com/sagernet/sing-box/constant"
|
|
"github.com/sagernet/sing-box/log"
|
|
"github.com/sagernet/sing-box/option"
|
|
"github.com/sagernet/sing-tun"
|
|
"github.com/sagernet/sing/common"
|
|
"github.com/sagernet/sing/common/batch"
|
|
E "github.com/sagernet/sing/common/exceptions"
|
|
M "github.com/sagernet/sing/common/metadata"
|
|
N "github.com/sagernet/sing/common/network"
|
|
"github.com/sagernet/sing/common/x/list"
|
|
"github.com/sagernet/sing/service"
|
|
"github.com/sagernet/sing/service/pause"
|
|
)
|
|
|
|
func RegisterURLTest(registry *outbound.Registry) {
|
|
outbound.Register[option.URLTestOutboundOptions](registry, C.TypeURLTest, NewURLTest)
|
|
}
|
|
|
|
var _ adapter.OutboundGroup = (*URLTest)(nil)
|
|
|
|
type URLTest struct {
|
|
outbound.Adapter
|
|
ctx context.Context
|
|
outbound adapter.OutboundManager
|
|
connection adapter.ConnectionManager
|
|
logger log.ContextLogger
|
|
tags []string
|
|
link string
|
|
interval time.Duration
|
|
tolerance uint16
|
|
idleTimeout time.Duration
|
|
group *URLTestGroup
|
|
interruptExternalConnections bool
|
|
}
|
|
|
|
func NewURLTest(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.URLTestOutboundOptions) (adapter.Outbound, error) {
|
|
outbound := &URLTest{
|
|
Adapter: outbound.NewAdapter(C.TypeURLTest, tag, []string{N.NetworkTCP, N.NetworkUDP}, options.Outbounds),
|
|
ctx: ctx,
|
|
outbound: service.FromContext[adapter.OutboundManager](ctx),
|
|
connection: service.FromContext[adapter.ConnectionManager](ctx),
|
|
logger: logger,
|
|
tags: options.Outbounds,
|
|
link: options.URL,
|
|
interval: time.Duration(options.Interval),
|
|
tolerance: options.Tolerance,
|
|
idleTimeout: time.Duration(options.IdleTimeout),
|
|
interruptExternalConnections: options.InterruptExistConnections,
|
|
}
|
|
if len(outbound.tags) == 0 {
|
|
return nil, E.New("missing tags")
|
|
}
|
|
return outbound, nil
|
|
}
|
|
|
|
func (s *URLTest) Start() error {
|
|
outbounds := make([]adapter.Outbound, 0, len(s.tags))
|
|
for i, tag := range s.tags {
|
|
detour, loaded := s.outbound.Outbound(tag)
|
|
if !loaded {
|
|
return E.New("outbound ", i, " not found: ", tag)
|
|
}
|
|
outbounds = append(outbounds, detour)
|
|
}
|
|
group, err := NewURLTestGroup(s.ctx, s.outbound, s.logger, outbounds, s.link, s.interval, s.tolerance, s.idleTimeout, s.interruptExternalConnections)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.group = group
|
|
return nil
|
|
}
|
|
|
|
func (s *URLTest) PostStart() error {
|
|
s.group.PostStart()
|
|
return nil
|
|
}
|
|
|
|
func (s *URLTest) Close() error {
|
|
return common.Close(
|
|
common.PtrOrNil(s.group),
|
|
)
|
|
}
|
|
|
|
func (s *URLTest) Now() string {
|
|
if s.group.selectedOutboundTCP != nil {
|
|
return s.group.selectedOutboundTCP.Tag()
|
|
} else if s.group.selectedOutboundUDP != nil {
|
|
return s.group.selectedOutboundUDP.Tag()
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func (s *URLTest) All() []string {
|
|
return s.tags
|
|
}
|
|
|
|
func (s *URLTest) URLTest(ctx context.Context) (map[string]uint16, error) {
|
|
return s.group.URLTest(ctx)
|
|
}
|
|
|
|
func (s *URLTest) CheckOutbounds() {
|
|
s.group.CheckOutbounds(true)
|
|
}
|
|
|
|
func (s *URLTest) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
|
|
s.group.Touch()
|
|
var outbound adapter.Outbound
|
|
switch N.NetworkName(network) {
|
|
case N.NetworkTCP:
|
|
outbound = s.group.selectedOutboundTCP
|
|
case N.NetworkUDP:
|
|
outbound = s.group.selectedOutboundUDP
|
|
default:
|
|
return nil, E.Extend(N.ErrUnknownNetwork, network)
|
|
}
|
|
if outbound == nil {
|
|
outbound, _ = s.group.Select(network)
|
|
}
|
|
if outbound == nil {
|
|
return nil, E.New("missing supported outbound")
|
|
}
|
|
conn, err := outbound.DialContext(ctx, network, destination)
|
|
if err == nil {
|
|
return s.group.interruptGroup.NewConn(conn, interrupt.IsExternalConnectionFromContext(ctx)), nil
|
|
}
|
|
s.logger.ErrorContext(ctx, err)
|
|
s.group.history.DeleteURLTestHistory(outbound.Tag())
|
|
return nil, err
|
|
}
|
|
|
|
func (s *URLTest) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) {
|
|
s.group.Touch()
|
|
outbound := s.group.selectedOutboundUDP
|
|
if outbound == nil {
|
|
outbound, _ = s.group.Select(N.NetworkUDP)
|
|
}
|
|
if outbound == nil {
|
|
return nil, E.New("missing supported outbound")
|
|
}
|
|
conn, err := outbound.ListenPacket(ctx, destination)
|
|
if err == nil {
|
|
return s.group.interruptGroup.NewPacketConn(conn, interrupt.IsExternalConnectionFromContext(ctx)), nil
|
|
}
|
|
s.logger.ErrorContext(ctx, err)
|
|
s.group.history.DeleteURLTestHistory(outbound.Tag())
|
|
return nil, err
|
|
}
|
|
|
|
func (s *URLTest) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
|
|
ctx = interrupt.ContextWithIsExternalConnection(ctx)
|
|
s.connection.NewConnection(ctx, s, conn, metadata, onClose)
|
|
}
|
|
|
|
func (s *URLTest) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
|
|
ctx = interrupt.ContextWithIsExternalConnection(ctx)
|
|
s.connection.NewPacketConnection(ctx, s, conn, metadata, onClose)
|
|
}
|
|
|
|
func (s *URLTest) NewDirectRouteConnection(metadata adapter.InboundContext, routeContext tun.DirectRouteContext, timeout time.Duration) (tun.DirectRouteDestination, error) {
|
|
s.group.Touch()
|
|
selected := s.group.selectedOutboundTCP
|
|
if selected == nil {
|
|
selected, _ = s.group.Select(N.NetworkTCP)
|
|
}
|
|
if selected == nil {
|
|
return nil, E.New("missing supported outbound")
|
|
}
|
|
if !common.Contains(selected.Network(), metadata.Network) {
|
|
return nil, E.New(metadata.Network, " is not supported by outbound: ", selected.Tag())
|
|
}
|
|
return selected.(adapter.DirectRouteOutbound).NewDirectRouteConnection(metadata, routeContext, timeout)
|
|
}
|
|
|
|
type URLTestGroup struct {
|
|
ctx context.Context
|
|
outbound adapter.OutboundManager
|
|
pause pause.Manager
|
|
pauseCallback *list.Element[pause.Callback]
|
|
logger log.Logger
|
|
outbounds []adapter.Outbound
|
|
link string
|
|
interval time.Duration
|
|
tolerance uint16
|
|
idleTimeout time.Duration
|
|
history *urltest.HistoryStorage
|
|
checking atomic.Bool
|
|
selectedOutboundTCP adapter.Outbound
|
|
selectedOutboundUDP adapter.Outbound
|
|
interruptGroup *interrupt.Group
|
|
interruptExternalConnections bool
|
|
access sync.Mutex
|
|
ticker *time.Ticker
|
|
close chan struct{}
|
|
started bool
|
|
lastActive common.TypedValue[time.Time]
|
|
}
|
|
|
|
func NewURLTestGroup(ctx context.Context, outboundManager adapter.OutboundManager, logger log.Logger, outbounds []adapter.Outbound, link string, interval time.Duration, tolerance uint16, idleTimeout time.Duration, interruptExternalConnections bool) (*URLTestGroup, error) {
|
|
if interval == 0 {
|
|
interval = C.DefaultURLTestInterval
|
|
}
|
|
if tolerance == 0 {
|
|
tolerance = 50
|
|
}
|
|
if idleTimeout == 0 {
|
|
idleTimeout = C.DefaultURLTestIdleTimeout
|
|
}
|
|
if interval > idleTimeout {
|
|
return nil, E.New("interval must be less or equal than idle_timeout")
|
|
}
|
|
history := service.PtrFromContext[urltest.HistoryStorage](ctx)
|
|
if history == nil {
|
|
return nil, E.New("missing URL test history storage")
|
|
}
|
|
return &URLTestGroup{
|
|
ctx: ctx,
|
|
outbound: outboundManager,
|
|
logger: logger,
|
|
outbounds: outbounds,
|
|
link: link,
|
|
interval: interval,
|
|
tolerance: tolerance,
|
|
idleTimeout: idleTimeout,
|
|
history: history,
|
|
close: make(chan struct{}),
|
|
pause: service.FromContext[pause.Manager](ctx),
|
|
interruptGroup: interrupt.NewGroup(),
|
|
interruptExternalConnections: interruptExternalConnections,
|
|
}, nil
|
|
}
|
|
|
|
func (g *URLTestGroup) PostStart() {
|
|
g.access.Lock()
|
|
defer g.access.Unlock()
|
|
g.started = true
|
|
g.lastActive.Store(time.Now())
|
|
go g.CheckOutbounds(false)
|
|
}
|
|
|
|
func (g *URLTestGroup) Touch() {
|
|
if !g.started {
|
|
return
|
|
}
|
|
g.access.Lock()
|
|
defer g.access.Unlock()
|
|
if g.ticker != nil {
|
|
g.lastActive.Store(time.Now())
|
|
return
|
|
}
|
|
ticker := time.NewTicker(g.interval)
|
|
g.ticker = ticker
|
|
g.pauseCallback = pause.RegisterTicker(g.pause, ticker, g.interval, nil)
|
|
go g.loopCheck(ticker, g.close)
|
|
}
|
|
|
|
func (g *URLTestGroup) Close() error {
|
|
g.access.Lock()
|
|
defer g.access.Unlock()
|
|
if g.ticker == nil {
|
|
return nil
|
|
}
|
|
g.ticker.Stop()
|
|
g.ticker = nil
|
|
g.pause.UnregisterCallback(g.pauseCallback)
|
|
g.pauseCallback = nil
|
|
close(g.close)
|
|
return nil
|
|
}
|
|
|
|
func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) {
|
|
var minDelay uint16
|
|
var minOutbound adapter.Outbound
|
|
switch network {
|
|
case N.NetworkTCP:
|
|
if g.selectedOutboundTCP != nil {
|
|
if history := g.history.LoadURLTestHistory(RealTag(g.selectedOutboundTCP)); history != nil {
|
|
minOutbound = g.selectedOutboundTCP
|
|
minDelay = history.Delay
|
|
}
|
|
}
|
|
case N.NetworkUDP:
|
|
if g.selectedOutboundUDP != nil {
|
|
if history := g.history.LoadURLTestHistory(RealTag(g.selectedOutboundUDP)); history != nil {
|
|
minOutbound = g.selectedOutboundUDP
|
|
minDelay = history.Delay
|
|
}
|
|
}
|
|
}
|
|
for _, detour := range g.outbounds {
|
|
if !common.Contains(detour.Network(), network) {
|
|
continue
|
|
}
|
|
history := g.history.LoadURLTestHistory(RealTag(detour))
|
|
if history == nil {
|
|
continue
|
|
}
|
|
if minDelay == 0 || minDelay > history.Delay+g.tolerance {
|
|
minDelay = history.Delay
|
|
minOutbound = detour
|
|
}
|
|
}
|
|
if minOutbound == nil {
|
|
for _, detour := range g.outbounds {
|
|
if !common.Contains(detour.Network(), network) {
|
|
continue
|
|
}
|
|
return detour, false
|
|
}
|
|
return nil, false
|
|
}
|
|
return minOutbound, true
|
|
}
|
|
|
|
func (g *URLTestGroup) loopCheck(ticker *time.Ticker, closeChan <-chan struct{}) {
|
|
if time.Since(g.lastActive.Load()) > g.interval {
|
|
g.lastActive.Store(time.Now())
|
|
g.CheckOutbounds(false)
|
|
}
|
|
for {
|
|
select {
|
|
case <-closeChan:
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
if time.Since(g.lastActive.Load()) > g.idleTimeout {
|
|
g.access.Lock()
|
|
if g.ticker == ticker {
|
|
g.ticker.Stop()
|
|
g.ticker = nil
|
|
g.pause.UnregisterCallback(g.pauseCallback)
|
|
g.pauseCallback = nil
|
|
}
|
|
g.access.Unlock()
|
|
return
|
|
}
|
|
g.CheckOutbounds(false)
|
|
}
|
|
}
|
|
|
|
func (g *URLTestGroup) CheckOutbounds(force bool) {
|
|
_, _ = g.urlTest(g.ctx, force)
|
|
}
|
|
|
|
func (g *URLTestGroup) URLTest(ctx context.Context) (map[string]uint16, error) {
|
|
return g.urlTest(ctx, false)
|
|
}
|
|
|
|
func (g *URLTestGroup) urlTest(ctx context.Context, force bool) (map[string]uint16, error) {
|
|
result := make(map[string]uint16)
|
|
if g.checking.Swap(true) {
|
|
return result, nil
|
|
}
|
|
defer g.checking.Store(false)
|
|
b, _ := batch.New(ctx, batch.WithConcurrencyNum[any](10))
|
|
checked := make(map[string]bool)
|
|
var resultAccess sync.Mutex
|
|
for _, detour := range g.outbounds {
|
|
tag := detour.Tag()
|
|
realTag := RealTag(detour)
|
|
if checked[realTag] {
|
|
continue
|
|
}
|
|
history := g.history.LoadURLTestHistory(realTag)
|
|
if !force && history != nil && time.Since(history.Time) < g.interval {
|
|
continue
|
|
}
|
|
checked[realTag] = true
|
|
p, loaded := g.outbound.Outbound(realTag)
|
|
if !loaded {
|
|
continue
|
|
}
|
|
b.Go(realTag, func() (any, error) {
|
|
testCtx, cancel := context.WithTimeout(g.ctx, C.TCPTimeout)
|
|
defer cancel()
|
|
t, err := urltest.URLTest(testCtx, g.link, p)
|
|
if err != nil {
|
|
g.logger.Debug("outbound ", tag, " unavailable: ", err)
|
|
g.history.DeleteURLTestHistory(realTag)
|
|
} else {
|
|
g.logger.Debug("outbound ", tag, " available: ", t, "ms")
|
|
g.history.StoreURLTestHistory(realTag, &adapter.URLTestHistory{
|
|
Time: time.Now(),
|
|
Delay: t,
|
|
})
|
|
resultAccess.Lock()
|
|
result[tag] = t
|
|
resultAccess.Unlock()
|
|
}
|
|
return nil, nil
|
|
})
|
|
}
|
|
b.Wait()
|
|
g.performUpdateCheck()
|
|
return result, nil
|
|
}
|
|
|
|
func (g *URLTestGroup) performUpdateCheck() {
|
|
var updated bool
|
|
if outbound, exists := g.Select(N.NetworkTCP); outbound != nil && (g.selectedOutboundTCP == nil || (exists && outbound != g.selectedOutboundTCP)) {
|
|
if g.selectedOutboundTCP != nil {
|
|
updated = true
|
|
}
|
|
g.selectedOutboundTCP = outbound
|
|
}
|
|
if outbound, exists := g.Select(N.NetworkUDP); outbound != nil && (g.selectedOutboundUDP == nil || (exists && outbound != g.selectedOutboundUDP)) {
|
|
if g.selectedOutboundUDP != nil {
|
|
updated = true
|
|
}
|
|
g.selectedOutboundUDP = outbound
|
|
}
|
|
if updated {
|
|
g.interruptGroup.Interrupt(g.interruptExternalConnections)
|
|
}
|
|
}
|