diff --git a/adapter/experimental.go b/adapter/experimental.go index 023e01bf..493d8027 100644 --- a/adapter/experimental.go +++ b/adapter/experimental.go @@ -168,10 +168,3 @@ type URLTestGroup interface { URLTest(ctx context.Context) (map[string]uint16, error) PerformUpdateCheck() } - -func OutboundTag(detour Outbound) string { - if group, isGroup := detour.(OutboundGroup); isGroup { - return group.Now() - } - return detour.Tag() -} diff --git a/daemon/started_service.go b/daemon/started_service.go index c088bead..0ca8e4bb 100644 --- a/daemon/started_service.go +++ b/daemon/started_service.go @@ -21,7 +21,6 @@ import ( "github.com/sagernet/sing-box/log" "github.com/sagernet/sing-box/protocol/group" "github.com/sagernet/sing/common" - "github.com/sagernet/sing/common/batch" "github.com/sagernet/sing/common/memory" "github.com/sagernet/sing/common/observable" "github.com/sagernet/sing/common/x/list" @@ -616,7 +615,7 @@ func (s *StartedService) readGroups() *Groups { var item GroupItem item.Tag = itemTag item.Type = itemOutbound.Type() - if history := historyStorage.LoadURLTestHistory(adapter.OutboundTag(itemOutbound)); history != nil { + if history := historyStorage.LoadURLTestHistory(group.RealTag(boxService.outboundManager, itemOutbound)); history != nil { item.UrlTestTime = history.Time.Unix() item.UrlTestDelay = int32(history.Delay) } @@ -729,33 +728,11 @@ func (s *StartedService) URLTest(ctx context.Context, request *URLTestRequest) ( if isURLTest { go urlTest.CheckOutbounds() } else if isOutboundGroup { - outbounds := common.Filter(common.Map(outboundGroup.All(), func(it string) adapter.Outbound { + outbounds := common.FilterNotNil(common.Map(outboundGroup.All(), func(it string) adapter.Outbound { itOutbound, _ := boxService.outboundManager.Outbound(it) return itOutbound - }), func(it adapter.Outbound) bool { - if it == nil { - return false - } - _, isGroup := it.(adapter.OutboundGroup) - return !isGroup - }) - b, _ := batch.New(boxService.ctx, batch.WithConcurrencyNum[any](10)) - for _, detour := range outbounds { - outboundToTest := detour - itemTag := outboundToTest.Tag() - b.Go(itemTag, func() (any, error) { - t, err := urltest.URLTest(boxService.ctx, "", outboundToTest) - if err != nil { - historyStorage.DeleteURLTestHistory(itemTag) - } else { - historyStorage.StoreURLTestHistory(itemTag, &adapter.URLTestHistory{ - Time: time.Now(), - Delay: t, - }) - } - return nil, nil - }) - } + })) + go group.URLTestOutbounds(boxService.ctx, boxService.outboundManager, historyStorage, boxService.logFactory.Logger(), outbounds, "", 0, true) } else { go func() { t, err := urltest.URLTest(boxService.ctx, "", outbound) @@ -1198,7 +1175,7 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server Tag: ob.Tag(), Type: ob.Type(), } - if history := historyStorage.LoadURLTestHistory(adapter.OutboundTag(ob)); history != nil { + if history := historyStorage.LoadURLTestHistory(group.RealTag(boxService.outboundManager, ob)); history != nil { item.UrlTestTime = history.Time.Unix() item.UrlTestDelay = int32(history.Delay) } @@ -1209,7 +1186,7 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server Tag: ep.Tag(), Type: ep.Type(), } - if history := historyStorage.LoadURLTestHistory(adapter.OutboundTag(ep)); history != nil { + if history := historyStorage.LoadURLTestHistory(group.RealTag(boxService.outboundManager, ep)); history != nil { item.UrlTestTime = history.Time.Unix() item.UrlTestDelay = int32(history.Delay) } diff --git a/experimental/clashapi/api_meta_group.go b/experimental/clashapi/api_meta_group.go index 31dbdaf6..725f3728 100644 --- a/experimental/clashapi/api_meta_group.go +++ b/experimental/clashapi/api_meta_group.go @@ -5,14 +5,11 @@ import ( "net/http" "strconv" "strings" - "sync" "time" "github.com/sagernet/sing-box/adapter" - "github.com/sagernet/sing-box/common/urltest" "github.com/sagernet/sing-box/protocol/group" "github.com/sagernet/sing/common" - "github.com/sagernet/sing/common/batch" "github.com/sagernet/sing/common/json/badjson" "github.com/go-chi/chi/v5" @@ -89,40 +86,7 @@ func getGroupDelay(server *Server) func(w http.ResponseWriter, r *http.Request) itOutbound, _ := server.outbound.Outbound(it) return itOutbound })) - b, _ := batch.New(ctx, batch.WithConcurrencyNum[any](10)) - checked := make(map[string]bool) - result = make(map[string]uint16) - var resultAccess sync.Mutex - for _, detour := range outbounds { - tag := detour.Tag() - realTag := group.RealTag(detour) - if checked[realTag] { - continue - } - checked[realTag] = true - p, loaded := server.outbound.Outbound(realTag) - if !loaded { - continue - } - b.Go(realTag, func() (any, error) { - t, err := urltest.URLTest(ctx, url, p) - if err != nil { - server.logger.Debug("outbound ", tag, " unavailable: ", err) - server.urlTestHistory.DeleteURLTestHistory(realTag) - } else { - server.logger.Debug("outbound ", tag, " available: ", t, "ms") - server.urlTestHistory.StoreURLTestHistory(realTag, &adapter.URLTestHistory{ - Time: time.Now(), - Delay: t, - }) - resultAccess.Lock() - result[tag] = t - resultAccess.Unlock() - } - return nil, nil - }) - } - b.Wait() + result = group.URLTestOutbounds(ctx, server.outbound, server.urlTestHistory, server.logger, outbounds, url, 0, true) } if err != nil { diff --git a/experimental/clashapi/proxies.go b/experimental/clashapi/proxies.go index afd7ba22..fcb159fa 100644 --- a/experimental/clashapi/proxies.go +++ b/experimental/clashapi/proxies.go @@ -70,7 +70,7 @@ func proxyInfo(server *Server, detour adapter.Outbound) *badjson.JSONObject { info.Put("type", clashType) info.Put("name", detour.Tag()) info.Put("udp", common.Contains(detour.Network(), N.NetworkUDP)) - delayHistory := server.urlTestHistory.LoadURLTestHistory(adapter.OutboundTag(detour)) + delayHistory := server.urlTestHistory.LoadURLTestHistory(group.RealTag(server.outbound, detour)) if delayHistory != nil { info.Put("history", []*adapter.URLTestHistory{delayHistory}) } else { @@ -193,7 +193,7 @@ func groupContains(outboundManager adapter.OutboundManager, outboundGroup adapte if !loaded { continue } - if group.RealTag(member) == tag { + if group.RealTag(outboundManager, member) == tag { return true } memberGroup, isGroup := member.(adapter.OutboundGroup) @@ -228,7 +228,7 @@ func getProxyDelay(server *Server) func(w http.ResponseWriter, r *http.Request) delay, err := urltest.URLTest(ctx, url, proxy) defer func() { - realTag := group.RealTag(proxy) + realTag := group.RealTag(server.outbound, proxy) if err != nil { server.urlTestHistory.DeleteURLTestHistory(realTag) } else { diff --git a/protocol/group/selector.go b/protocol/group/selector.go index 2b1cb040..0542d073 100644 --- a/protocol/group/selector.go +++ b/protocol/group/selector.go @@ -180,9 +180,18 @@ func (s *Selector) NewPacketConnection(ctx context.Context, conn N.PacketConn, m } } -func RealTag(detour adapter.Outbound) string { - if group, isGroup := detour.(adapter.OutboundGroup); isGroup { - return group.Now() +func RealTag(outboundManager adapter.OutboundManager, detour adapter.Outbound) string { + tag := detour.Tag() + for { + group, isGroup := detour.(adapter.OutboundGroup) + if !isGroup { + return tag + } + tag = group.Now() + var loaded bool + detour, loaded = outboundManager.Outbound(tag) + if !loaded { + return tag + } } - return detour.Tag() } diff --git a/protocol/group/urltest.go b/protocol/group/urltest.go index 5dfa2440..ba3197fb 100644 --- a/protocol/group/urltest.go +++ b/protocol/group/urltest.go @@ -2,6 +2,7 @@ package group import ( "context" + "maps" "net" "sync" "sync/atomic" @@ -297,14 +298,14 @@ func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) { switch network { case N.NetworkTCP: if g.selectedOutboundTCP != nil { - if history := g.history.LoadURLTestHistory(RealTag(g.selectedOutboundTCP)); history != nil { + if history := g.history.LoadURLTestHistory(RealTag(g.outbound, 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 { + if history := g.history.LoadURLTestHistory(RealTag(g.outbound, g.selectedOutboundUDP)); history != nil { minOutbound = g.selectedOutboundUDP minDelay = history.Delay } @@ -314,7 +315,7 @@ func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) { if !common.Contains(detour.Network(), network) { continue } - history := g.history.LoadURLTestHistory(RealTag(detour)) + history := g.history.LoadURLTestHistory(RealTag(g.outbound, detour)) if history == nil { continue } @@ -366,7 +367,17 @@ func (g *URLTestGroup) CheckOutbounds(ctx context.Context, force bool) { } func (g *URLTestGroup) URLTest(ctx context.Context) (map[string]uint16, error) { - return g.urlTest(ctx, false) + return g.urlTest(ctx, true) +} + +func (g *URLTestGroup) urlTest(ctx context.Context, force bool) (map[string]uint16, error) { + if g.checking.Swap(true) { + return make(map[string]uint16), nil + } + defer g.checking.Store(false) + result := URLTestOutbounds(ctx, g.outbound, g.history, g.logger, g.outbounds, g.link, g.interval, force) + g.performUpdateCheck() + return result, nil } type urlTestResult struct { @@ -374,63 +385,101 @@ type urlTestResult struct { err error } -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) +type urlTestBatch struct { + ctx context.Context + outbound adapter.OutboundManager + history *urltest.HistoryStorage + logger log.Logger + batch *batch.Batch[any] + checked map[string]bool + groups []adapter.OutboundGroup + access sync.Mutex + result map[string]uint16 +} + +func URLTestOutbounds(ctx context.Context, outboundManager adapter.OutboundManager, history *urltest.HistoryStorage, logger log.Logger, outbounds []adapter.Outbound, link string, interval time.Duration, force bool) map[string]uint16 { 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(ctx, C.TCPTimeout) - defer cancel() - testChan := make(chan urlTestResult, 1) - go func() { - delay, testErr := urltest.URLTest(testCtx, g.link, p) - testChan <- urlTestResult{delay, testErr} - }() - var testResult urlTestResult - select { - case testResult = <-testChan: - case <-testCtx.Done(): - testResult.err = testCtx.Err() - } - if testResult.err != nil { - g.logger.Debug("outbound ", tag, " unavailable: ", testResult.err) - g.history.DeleteURLTestHistory(realTag) - } else { - g.logger.Debug("outbound ", tag, " available: ", testResult.delay, "ms") - g.history.StoreURLTestHistory(realTag, &adapter.URLTestHistory{ - Time: time.Now(), - Delay: testResult.delay, - }) - resultAccess.Lock() - result[tag] = testResult.delay - resultAccess.Unlock() - } - return nil, nil - }) + testBatch := &urlTestBatch{ + ctx: ctx, + outbound: outboundManager, + history: history, + logger: logger, + batch: b, + checked: make(map[string]bool), + result: make(map[string]uint16), } + testBatch.test(outbounds, link, interval, force) b.Wait() - g.performUpdateCheck() - return result, nil + for _, outboundGroup := range testBatch.groups { + groupHistory := history.LoadURLTestHistory(RealTag(outboundManager, outboundGroup)) + if groupHistory != nil { + testBatch.result[outboundGroup.Tag()] = groupHistory.Delay + } + } + return testBatch.result +} + +func (b *urlTestBatch) test(outbounds []adapter.Outbound, link string, interval time.Duration, force bool) { + for _, detour := range outbounds { + tag := detour.Tag() + if b.checked[tag] { + continue + } + switch nested := detour.(type) { + case *URLTest: + b.checked[tag] = true + b.groups = append(b.groups, nested) + b.batch.Go(tag, func() (any, error) { + nestedResult, _ := nested.group.urlTest(b.ctx, force) + b.access.Lock() + maps.Copy(b.result, nestedResult) + b.access.Unlock() + return nil, nil + }) + case adapter.OutboundGroup: + b.checked[tag] = true + b.groups = append(b.groups, nested) + b.test(common.FilterNotNil(common.Map(nested.All(), func(it string) adapter.Outbound { + member, _ := b.outbound.Outbound(it) + return member + })), link, interval, force) + default: + history := b.history.LoadURLTestHistory(tag) + if !force && history != nil && time.Since(history.Time) < interval { + continue + } + b.checked[tag] = true + b.batch.Go(tag, func() (any, error) { + testCtx, cancel := context.WithTimeout(b.ctx, C.TCPTimeout) + defer cancel() + testChan := make(chan urlTestResult, 1) + go func() { + delay, testErr := urltest.URLTest(testCtx, link, detour) + testChan <- urlTestResult{delay, testErr} + }() + var testResult urlTestResult + select { + case testResult = <-testChan: + case <-testCtx.Done(): + testResult.err = testCtx.Err() + } + if testResult.err != nil { + b.logger.Debug("outbound ", tag, " unavailable: ", testResult.err) + b.history.DeleteURLTestHistory(tag) + } else { + b.logger.Debug("outbound ", tag, " available: ", testResult.delay, "ms") + b.history.StoreURLTestHistory(tag, &adapter.URLTestHistory{ + Time: time.Now(), + Delay: testResult.delay, + }) + b.access.Lock() + b.result[tag] = testResult.delay + b.access.Unlock() + } + return nil, nil + }) + } + } } func (g *URLTestGroup) performUpdateCheck() {