From 103497ff69729d7372b46430124d3d0fc4aa8959 Mon Sep 17 00:00:00 2001 From: zhboner Date: Wed, 15 Apr 2026 14:48:57 +0800 Subject: [PATCH] feat: smart group implemented. --- adapter/adapter.go | 103 ++++- adapter/outboundgroup/fallback.go | 4 + adapter/outboundgroup/groupbase.go | 17 +- adapter/outboundgroup/parser.go | 98 ++++ adapter/outboundgroup/selector.go | 4 + adapter/outboundgroup/smart.go | 720 +++++++++++++++++++++++++++++ adapter/outboundgroup/urltest.go | 5 + constant/adapters.go | 16 + tunnel/statistic/manager.go | 16 + tunnel/statistic/speedobserver.go | 134 ++++++ tunnel/tunnel.go | 12 + 11 files changed, 1120 insertions(+), 9 deletions(-) create mode 100644 adapter/outboundgroup/smart.go create mode 100644 tunnel/statistic/speedobserver.go diff --git a/adapter/adapter.go b/adapter/adapter.go index 6f20c08abb..fb635b6d79 100644 --- a/adapter/adapter.go +++ b/adapter/adapter.go @@ -4,9 +4,11 @@ import ( "context" "encoding/json" "fmt" + "math" "net" "net/url" "strings" + "sync" "time" "github.com/metacubex/mihomo/common/atomic" @@ -23,7 +25,11 @@ import ( var UnifiedDelay = atomic.NewBool(false) const ( - defaultHistoriesNum = 10 + defaultHistoriesNum = 10 + defaultSpeedHistoriesNum = 120 + defaultTestResultsNum = 20 + speedStaleThreshold = 30 * time.Second + speedDecayLambda = 0.001 ) type internalProxyState struct { @@ -36,6 +42,22 @@ type Proxy struct { alive atomic.Bool history *queue.Queue[C.DelayHistory] extra xsync.Map[string, *internalProxyState] + + peakSpeed atomic.Uint64 + peakAt atomic.Int64 + currentSpeed atomic.Uint64 + currentAt atomic.Int64 + + speedHistoryMu sync.Mutex + speedHistory []C.SpeedHistory + + testResultsMu sync.Mutex + testResults []TestResult +} + +type TestResult struct { + At time.Time + Success bool } // Adapter implements C.Proxy @@ -286,6 +308,85 @@ func NewProxy(adapter C.ProxyAdapter) *Proxy { } } +// PushSpeed implements C.Proxy - updates current speed and peak if exceeds decayed peak +func (p *Proxy) PushSpeed(speed uint64) { + if speed == 0 { + return + } + now := time.Now() + p.currentSpeed.Store(speed) + p.currentAt.Store(now.UnixNano()) + + peak := p.peakSpeed.Load() + peakAt := time.Unix(0, p.peakAt.Load()) + decayedPeak := uint64(float64(peak) * math.Exp(-speedDecayLambda*now.Sub(peakAt).Seconds())) + if speed >= decayedPeak { + p.peakSpeed.Store(speed) + p.peakAt.Store(now.UnixNano()) + } + + p.speedHistoryMu.Lock() + p.speedHistory = append(p.speedHistory, C.SpeedHistory{Time: now, Speed: speed}) + if len(p.speedHistory) > defaultSpeedHistoriesNum { + p.speedHistory = p.speedHistory[1:] + } + p.speedHistoryMu.Unlock() +} + +// LastSpeed implements C.Proxy - returns fresh speed or 0 if stale +func (p *Proxy) LastSpeed() uint64 { + at := time.Unix(0, p.currentAt.Load()) + if at.IsZero() || time.Since(at) > speedStaleThreshold { + return 0 + } + return p.currentSpeed.Load() +} + +// EffectiveSpeed implements C.Proxy - max(decayed peak, fresh current) +func (p *Proxy) EffectiveSpeed() uint64 { + current := p.LastSpeed() + peak := uint64(float64(p.peakSpeed.Load()) * math.Exp(-speedDecayLambda*time.Since(time.Unix(0, p.peakAt.Load())).Seconds())) + if peak > current { + return peak + } + return current +} + +// SpeedHistory implements C.Proxy +func (p *Proxy) SpeedHistory() []C.SpeedHistory { + p.speedHistoryMu.Lock() + out := make([]C.SpeedHistory, len(p.speedHistory)) + copy(out, p.speedHistory) + p.speedHistoryMu.Unlock() + return out +} + +// PushTestResult implements C.Proxy - record test result for packet loss calculation +func (p *Proxy) PushTestResult(url string, success bool) { + p.testResultsMu.Lock() + p.testResults = append(p.testResults, TestResult{At: time.Now(), Success: success}) + if len(p.testResults) > defaultTestResultsNum { + p.testResults = p.testResults[1:] + } + p.testResultsMu.Unlock() +} + +// PacketLossRate implements C.Proxy - calculate loss rate from recent test results +func (p *Proxy) PacketLossRate(url string) float64 { + p.testResultsMu.Lock() + defer p.testResultsMu.Unlock() + if len(p.testResults) == 0 { + return 0 + } + fails := 0 + for _, r := range p.testResults { + if !r.Success { + fails++ + } + } + return float64(fails) / float64(len(p.testResults)) +} + func urlToMetadata(rawURL string) (addr C.Metadata, err error) { u, err := url.Parse(rawURL) if err != nil { diff --git a/adapter/outboundgroup/fallback.go b/adapter/outboundgroup/fallback.go index ef4a74c37f..79440bb54e 100644 --- a/adapter/outboundgroup/fallback.go +++ b/adapter/outboundgroup/fallback.go @@ -99,6 +99,10 @@ func (f *Fallback) Unwrap(metadata *C.Metadata, touch bool) C.Proxy { return proxy } +func (f *Fallback) EffectiveSpeed() uint64 { + return f.findAliveProxy(false).EffectiveSpeed() +} + func (f *Fallback) findAliveProxy(touch bool) C.Proxy { proxies := f.GetProxies(touch) for _, proxy := range proxies { diff --git a/adapter/outboundgroup/groupbase.go b/adapter/outboundgroup/groupbase.go index 7dd085c033..d26f772a34 100644 --- a/adapter/outboundgroup/groupbase.go +++ b/adapter/outboundgroup/groupbase.go @@ -233,19 +233,20 @@ func (gb *GroupBase) URLTest(ctx context.Context, url string, expectedStatus uti var lock sync.Mutex mp := map[string]uint16{} proxies := gb.GetProxies(false) - for _, proxy := range proxies { - proxy := proxy + for _, p := range proxies { + if p == nil { + continue + } wg.Add(1) - go func() { - delay, err := proxy.URLTest(ctx, url, expectedStatus) + go func(p C.Proxy) { + defer wg.Done() + delay, err := p.URLTest(ctx, url, expectedStatus) if err == nil { lock.Lock() - mp[proxy.Name()] = delay + mp[p.Name()] = delay lock.Unlock() } - - wg.Done() - }() + }(p) } wg.Wait() diff --git a/adapter/outboundgroup/parser.go b/adapter/outboundgroup/parser.go index d5ce72888e..4b233d5220 100644 --- a/adapter/outboundgroup/parser.go +++ b/adapter/outboundgroup/parser.go @@ -3,6 +3,7 @@ package outboundgroup import ( "errors" "fmt" + "math" "strings" "github.com/dlclark/regexp2" @@ -184,6 +185,12 @@ func ParseProxyGroup(config map[string]any, proxyMap map[string]C.Proxy, provide case "load-balance": strategy := parseStrategy(config) return NewLoadBalance(groupOption, providers, strategy) + case "smart": + opts, err := parseSmartOption(config) + if err != nil { + return nil, fmt.Errorf("%s: %w", groupName, err) + } + return NewSmart(groupOption, providers, opts...) case "relay": return nil, fmt.Errorf("%w: The group [%s] with relay type was removed, please using dialer-proxy instead", errType, groupName) default: @@ -230,3 +237,94 @@ func addTestUrlToProviders(providers []P.ProxyProvider, url string, expectedStat pd.RegisterHealthCheckTask(url, expectedStatus, filter, interval) } } + +func parseSmartOption(config map[string]any) ([]smartOption, error) { + opts := []smartOption{} + + weightDelay := defaultWeightDelay + weightLoss := defaultWeightLoss + weightSpeed := defaultWeightSpeed + tolerance := defaultTolerance + + if v, ok := config["weight-delay"]; ok { + if f, ok := parseNumeric(v); ok { + if f < 0 || math.IsNaN(f) || math.IsInf(f, 0) { + return nil, fmt.Errorf("weight-delay must be >= 0") + } + weightDelay = f + } + } + if v, ok := config["weight-loss"]; ok { + if f, ok := parseNumeric(v); ok { + if f < 0 || math.IsNaN(f) || math.IsInf(f, 0) { + return nil, fmt.Errorf("weight-loss must be >= 0") + } + weightLoss = f + } + } + if v, ok := config["weight-speed"]; ok { + if f, ok := parseNumeric(v); ok { + if f < 0 || math.IsNaN(f) || math.IsInf(f, 0) { + return nil, fmt.Errorf("weight-speed must be >= 0") + } + weightSpeed = f + } + } + + sum := weightDelay + weightLoss + weightSpeed + if sum == 0 { + return nil, fmt.Errorf("at least one weight must be > 0") + } + weightDelay /= sum + weightLoss /= sum + weightSpeed /= sum + + if v, ok := config["tolerance"]; ok { + if f, ok := parseNumeric(v); ok { + if f < 0 { + return nil, fmt.Errorf("tolerance must be >= 0") + } + tolerance = f + } + } + + opts = append(opts, func(s *Smart) { + s.weightDelay = weightDelay + s.weightLoss = weightLoss + s.weightSpeed = weightSpeed + s.tolerance = tolerance + }) + + return opts, nil +} + +func parseNumeric(v any) (float64, bool) { + switch x := v.(type) { + case float64: + return x, true + case float32: + return float64(x), true + case int: + return float64(x), true + case int8: + return float64(x), true + case int16: + return float64(x), true + case int32: + return float64(x), true + case int64: + return float64(x), true + case uint: + return float64(x), true + case uint8: + return float64(x), true + case uint16: + return float64(x), true + case uint32: + return float64(x), true + case uint64: + return float64(x), true + default: + return 0, false + } +} diff --git a/adapter/outboundgroup/selector.go b/adapter/outboundgroup/selector.go index b0272f6c1a..a5bc0c4b8a 100644 --- a/adapter/outboundgroup/selector.go +++ b/adapter/outboundgroup/selector.go @@ -95,6 +95,10 @@ func (s *Selector) Unwrap(metadata *C.Metadata, touch bool) C.Proxy { return s.selectedProxy(touch) } +func (s *Selector) EffectiveSpeed() uint64 { + return s.selectedProxy(false).EffectiveSpeed() +} + func (s *Selector) selectedProxy(touch bool) C.Proxy { proxies := s.GetProxies(touch) for _, proxy := range proxies { diff --git a/adapter/outboundgroup/smart.go b/adapter/outboundgroup/smart.go new file mode 100644 index 0000000000..f9d54d702b --- /dev/null +++ b/adapter/outboundgroup/smart.go @@ -0,0 +1,720 @@ +package outboundgroup + +import ( + "context" + "encoding/json" + "errors" + "math" + "math/rand" + "sync" + "time" + + "github.com/metacubex/mihomo/common/callback" + N "github.com/metacubex/mihomo/common/net" + "github.com/metacubex/mihomo/common/singledo" + "github.com/metacubex/mihomo/common/utils" + "github.com/metacubex/mihomo/common/xsync" + C "github.com/metacubex/mihomo/constant" + P "github.com/metacubex/mihomo/constant/provider" +) + +const ( + // 速度归一化参考值（P90 锚定） + speedRefFloor = 4 * 1024 * 1024 // 4 MiB/s，样本不足时的地板值，新节点达到此速度即可拿满分 + speedRefCeiling = 32 * 1024 * 1024 // 32 MiB/s，P90 上限，防止快网络中速度差异被过度放大 + minSpeedSamples = 3 // 计算 P90 所需的最小速度样本数，不足时用 floor 保底 + + // EMA 基线更新 + baselineAlpha = 0.20 // 指数移动平均系数，新样本权重 20%，历史权重 80% + baselineMinSamples = 3 // 建立基线所需的最小成功样本数，此前用算术平均 bootstrap + spikeIgnoreFactor = 3.0 // 异常值忽略因子，延迟 > 基线×3 视为 spike 不更新基线 + baselineMaxStepUp = 1.25 // 基线单次最大上涨比例 25%，防止 spike 缓慢污染基线 + reseedHighStreak = 5 // 连续高延迟次数触发中位数重校准，说明网络环境真的变了 + + // 退化检测与恢复 + degradeFactor = 1.50 // 退化因子，延迟 > 基线×1.5 标记退化 + recoverFactor = 1.20 // 恢复因子，延迟 ≤ 基线×1.2 才可清除退化（滞后防抖） + recoverSuccesses = 2 // 清除退化所需连续成功次数 + degradePenalty = 0.5 // 退化节点评分惩罚值（加到最终 score 上） + + // 评分权重（默认值，可在 YAML 中覆盖） + defaultWeightDelay = 0.3 // 延迟权重 + defaultWeightLoss = 0.3 // 丢包率权重 + defaultWeightSpeed = 0.4 // 速度权重 + + // 其他默认参数 + defaultTolerance = 0.1 // 评分切换容差，当前节点评分 ≤ 最优节点评分 + tolerance 时不切换 + defaultDegradeFactor = 1.5 // 退化因子默认值 + defaultDecayLambda = 0.001 // 速度衰减系数，半衰期 ≈ 11.5 分钟 +) + +type smartOption func(*Smart) + +type nodeMetrics struct { + proxy C.Proxy + delay uint16 + loss float64 + speed uint64 + alive bool +} + +type TestResult struct { + At time.Time + Success bool +} + +type NodeState struct { + mu sync.Mutex + LastObservedAt time.Time + Baseline uint64 + BaselineSamples int + HighLatencyStreak int + RecentSuccessDelays []uint16 + Degraded bool + RecoverStreak int + TestResults []TestResult +} + +type Smart struct { + *GroupBase + selected string + testUrl string + expectedStatus string + tolerance float64 + disableUDP bool + fastNode C.Proxy + fastSingle *singledo.Single[C.Proxy] + + weightDelay float64 + weightLoss float64 + weightSpeed float64 + + nodeStates xsync.Map[string, *NodeState] +} + +func (s *Smart) Now() string { + return s.fast(false).Name() +} + +func (s *Smart) Set(name string) error { + var p C.Proxy + for _, proxy := range s.GetProxies(false) { + if proxy.Name() == name { + p = proxy + break + } + } + if p == nil { + return errors.New("proxy not exist") + } + s.ForceSet(name) + return nil +} + +func (s *Smart) ForceSet(name string) { + s.selected = name + s.fastSingle.Reset() +} + +func (s *Smart) DialContext(ctx context.Context, metadata *C.Metadata) (c C.Conn, err error) { + proxy := s.fast(true) + c, err = proxy.DialContext(ctx, metadata) + if err == nil { + c.AppendToChains(s) + } else { + s.onDialFailed(proxy, err) + } + + if N.NeedHandshake(c) { + c = callback.NewFirstWriteCallBackConn(c, func(err error) { + if err == nil { + s.onDialSuccess() + } else { + s.onDialFailed(proxy, err) + } + }) + } + + return c, err +} + +func (s *Smart) ListenPacketContext(ctx context.Context, metadata *C.Metadata) (C.PacketConn, error) { + proxy := s.fast(true) + pc, err := proxy.ListenPacketContext(ctx, metadata) + if err == nil { + pc.AppendToChains(s) + } else { + s.onDialFailed(proxy, err) + } + + return pc, err +} + +func (s *Smart) SupportUDP() bool { + if s.disableUDP { + return false + } + return s.fast(false).SupportUDP() +} + +func (s *Smart) IsL3Protocol(metadata *C.Metadata) bool { + return s.fast(false).IsL3Protocol(metadata) +} + +func (s *Smart) Unwrap(metadata *C.Metadata, touch bool) C.Proxy { + return s.fast(touch) +} + +func (s *Smart) MarshalJSON() ([]byte, error) { + all := []string{} + for _, proxy := range s.GetProxies(false) { + all = append(all, proxy.Name()) + } + return json.Marshal(map[string]any{ + "type": s.Type().String(), + "now": s.Now(), + "all": all, + "testUrl": s.testUrl, + "expectedStatus": s.expectedStatus, + "hidden": s.Hidden(), + "icon": s.Icon(), + "weight-delay": s.weightDelay, + "weight-loss": s.weightLoss, + "weight-speed": s.weightSpeed, + "tolerance": s.tolerance, + }) +} + +func (s *Smart) Providers() []P.ProxyProvider { + return s.providers +} + +func (s *Smart) Proxies() []C.Proxy { + return s.GetProxies(false) +} + +func (s *Smart) URLTest(ctx context.Context, url string, expectedStatus utils.IntRanges[uint16]) (map[string]uint16, error) { + return s.GroupBase.URLTest(ctx, s.testUrl, expectedStatus) +} + +func (s *Smart) resolve(proxy C.Proxy) C.Proxy { + // 通过 proxy.Adapter() 获取底层 adapter（因为候选节点可能是 *adapter.Proxy 包装） + // type switch 到具体 group 类型，调用其 fast/findAliveProxy/selectedProxy 拿到 leaf + // leaf proxy 直接返回，因为速度/延迟/丢包等物理指标只在 leaf 层有意义 + adapter := proxy.Adapter() + switch a := adapter.(type) { + case *URLTest: + return a.fast(false) + case *Fallback: + return a.findAliveProxy(false) + case *Selector: + return a.selectedProxy(false) + case *Smart: + return a.fast(false) + default: + return proxy + } +} + +func (s *Smart) fast(touch bool) C.Proxy { + elm, _, _ := s.fastSingle.Do(func() (C.Proxy, error) { + proxies := s.GetProxies(touch) + if len(proxies) == 0 { + return nil, errors.New("no proxies") + } + + if s.selected != "" { + for _, p := range proxies { + if !p.AliveForTestUrl(s.testUrl) { + continue + } + if p.Name() == s.selected { + s.fastNode = p + return p, nil + } + } + } + + var aliveProxies []C.Proxy + hasData := false + for _, p := range proxies { + if p.AliveForTestUrl(s.testUrl) { + aliveProxies = append(aliveProxies, p) + } + if p.LastDelayForTestUrl(s.testUrl) != 0xffff { + hasData = true + } + } + if len(aliveProxies) == 0 { + chosen := proxies[rand.Intn(len(proxies))] + s.fastNode = chosen + return chosen, nil + } + if !hasData { + chosen := aliveProxies[rand.Intn(len(aliveProxies))] + s.fastNode = chosen + return chosen, nil + } + + // 收集所有候选节点指标（通过 resolve 拿到 leaf proxy 读取物理指标） + metrics := make([]nodeMetrics, 0, len(proxies)) + for _, p := range proxies { + resolved := s.resolve(p) + s.ingestProbeResult(p.Name(), resolved) + metrics = append(metrics, nodeMetrics{ + proxy: p, + delay: resolved.LastDelayForTestUrl(s.testUrl), + loss: s.PacketLossRate(p.Name()), + speed: resolved.EffectiveSpeed(), + alive: resolved.AliveForTestUrl(s.testUrl), + }) + } + + best := s.selectBest(metrics) + s.fastNode = best + return best, nil + }) + return elm +} + +func (s *Smart) ingestProbeResult(proxyName string, resolved C.Proxy) { + extra := resolved.ExtraDelayHistories() + state, ok := extra[s.testUrl] + if !ok || len(state.History) == 0 { + return + } + + latest := state.History[len(state.History)-1] + nodeState := s.getNodeState(proxyName) + nodeState.mu.Lock() + if !latest.Time.After(nodeState.LastObservedAt) { + nodeState.mu.Unlock() + return + } + nodeState.LastObservedAt = latest.Time + nodeState.mu.Unlock() + + success := latest.Delay > 0 && state.Alive + s.PushTestResult(proxyName, success) + s.updateBaseline(proxyName, success, latest.Delay) + s.updateDegraded(proxyName, success, latest.Delay) +} + +func (s *Smart) selectBest(metrics []nodeMetrics) C.Proxy { + // 1. 计算速度参考值（P90 锚定，限制在 [floor, ceiling]） + speedRef := s.computeSpeedRef(metrics) + var best C.Proxy + var bestScore float64 = math.MaxFloat64 + var currentScore float64 = math.MaxFloat64 + + // 2. 遍历所有存活节点，计算加权评分，选最低分 + for _, m := range metrics { + if !m.alive { + continue + } + score := s.calculateScore(m, speedRef) + if score < bestScore { + bestScore = score + best = m.proxy + } + // 记录当前节点的评分 + if s.fastNode != nil && m.proxy == s.fastNode { + currentScore = score + } + } + + // 3. 兜底：如果所有节点都不存活，选第一个 + if best == nil && len(metrics) > 0 { + best = metrics[0].proxy + } + + // 4. 评分容差防抖：当前节点评分 ≤ 最优节点评分 + tolerance 时不切换 + // 维度一致（都是评分），只有评分优势超过 tolerance 才切换 + if s.fastNode != nil && currentScore <= bestScore+s.tolerance { + return s.fastNode + } + + return best +} + +func (s *Smart) computeSpeedRef(metrics []nodeMetrics) float64 { + // 收集所有存活且有速度数据的节点 + speeds := make([]float64, 0, len(metrics)) + for _, m := range metrics { + if m.alive && m.speed > 0 { + speeds = append(speeds, float64(m.speed)) + } + } + // 样本不足 3 个时用地板值保底，避免 0/0 和慢网络 inflation + if len(speeds) < minSpeedSamples { + return speedRefFloor + } + // P90 锚定：限制在 [floor, ceiling] 防止极端值 + p90 := percentile(speeds, 0.90) + if p90 < speedRefFloor { + return speedRefFloor + } + if p90 > speedRefCeiling { + return speedRefCeiling + } + return p90 +} + +func (s *Smart) calculateScore(m nodeMetrics, speedRef float64) float64 { + // 延迟归一化：delay/2000ms，上限 1.0（2000ms 以上视为最差） + normDelay := float64(m.delay) / 2000.0 + if normDelay > 1.0 { + normDelay = 1.0 + } + // 丢包归一化：直接使用失败率，上限 1.0 + normLoss := m.loss + if normLoss > 1.0 { + normLoss = 1.0 + } + + var score float64 + if m.speed == 0 { + // 无速度数据时跳过速度项（等价于 norm_speed=1，中性值不奖不罚） + score = s.weightDelay*normDelay + s.weightLoss*normLoss + } else { + // 速度归一化：speed/speedRef，上限 1.0 + normSpeed := float64(m.speed) / speedRef + if normSpeed > 1.0 { + normSpeed = 1.0 + } + // 速度是正向指标，所以用 (1 - normSpeed) + score = s.weightDelay*normDelay + s.weightLoss*normLoss + s.weightSpeed*(1.0-normSpeed) + } + + // 退化节点加惩罚分 + if s.isDegraded(m.proxy.Name()) { + score += degradePenalty + } + + return score +} + +func (s *Smart) getNodeState(name string) *NodeState { + state, _ := s.nodeStates.LoadOrStoreFn(name, func() *NodeState { + return &NodeState{ + TestResults: make([]TestResult, 0, 20), + RecentSuccessDelays: make([]uint16, 0, 10), + } + }) + return state +} + +func (s *Smart) updateBaseline(proxyName string, success bool, delay uint16) { + state := s.getNodeState(proxyName) + state.mu.Lock() + defer state.mu.Unlock() + + if !success { + return + } + + // 样本不足时，用算术平均 bootstrap 基线 + if state.BaselineSamples < baselineMinSamples { + state.BaselineSamples++ + state.Baseline = (state.Baseline*uint64(state.BaselineSamples-1) + uint64(delay)) / uint64(state.BaselineSamples) + state.RecentSuccessDelays = append(state.RecentSuccessDelays, delay) + return + } + + // 异常值忽略：延迟 > 基线×3 视为 spike，不更新基线 + // 连续 5 次高延迟说明网络环境真的变了，用中位数重校准 + if uint64(delay) > uint64(float64(state.Baseline)*spikeIgnoreFactor) { + state.HighLatencyStreak++ + if state.HighLatencyStreak >= reseedHighStreak { + state.Baseline = medianUint16(state.RecentSuccessDelays) + state.HighLatencyStreak = 0 + } + return + } + + state.HighLatencyStreak = 0 + state.RecentSuccessDelays = append(state.RecentSuccessDelays, delay) + if len(state.RecentSuccessDelays) > 10 { + state.RecentSuccessDelays = state.RecentSuccessDelays[1:] + } + + // Capped EMA：单次最多上涨 25%，防止 spike 缓慢污染基线 + capped := uint64(delay) + maxUp := uint64(float64(state.Baseline) * baselineMaxStepUp) + if capped > maxUp { + capped = maxUp + } + state.Baseline = uint64((1.0-baselineAlpha)*float64(state.Baseline) + baselineAlpha*float64(capped)) +} + +func (s *Smart) updateDegraded(proxyName string, success bool, delay uint16) { + state := s.getNodeState(proxyName) + state.mu.Lock() + defer state.mu.Unlock() + + // 失败立即标记退化 + if !success { + state.Degraded = true + state.RecoverStreak = 0 + return + } + + // 基线未建立时，连续 2 次成功清除退化 + if state.BaselineSamples < 3 { + if state.Degraded { + state.RecoverStreak++ + if state.RecoverStreak >= recoverSuccesses { + state.Degraded = false + state.RecoverStreak = 0 + } + } + return + } + + // 延迟 > 基线×degradeFactor → 标记退化 + if uint64(delay) > uint64(float64(state.Baseline)*degradeFactor) { + state.Degraded = true + state.RecoverStreak = 0 + return + } + + // 滞后清除：延迟 ≤ 基线×recoverFactor 且连续 N 次成功才清除 + // recoverFactor < degradeFactor 防止抖动（hysteresis） + if state.Degraded && uint64(delay) <= uint64(float64(state.Baseline)*recoverFactor) { + state.RecoverStreak++ + if state.RecoverStreak >= recoverSuccesses { + state.Degraded = false + state.RecoverStreak = 0 + } + } else { + state.RecoverStreak = 0 + } +} + +func (s *Smart) markDegraded(proxyName string) { + state := s.getNodeState(proxyName) + state.mu.Lock() + defer state.mu.Unlock() + state.Degraded = true + state.RecoverStreak = 0 +} + +func (s *Smart) isDegraded(proxyName string) bool { + state, ok := s.nodeStates.Load(proxyName) + if !ok { + return false + } + state.mu.Lock() + defer state.mu.Unlock() + return state.Degraded +} + +func (s *Smart) PushTestResult(proxyName string, success bool) { + state := s.getNodeState(proxyName) + state.mu.Lock() + defer state.mu.Unlock() + state.TestResults = append(state.TestResults, TestResult{At: time.Now(), Success: success}) + if len(state.TestResults) > 20 { + state.TestResults = state.TestResults[1:] + } +} + +func (s *Smart) PacketLossRate(proxyName string) float64 { + state, ok := s.nodeStates.Load(proxyName) + if !ok { + return 0 + } + state.mu.Lock() + defer state.mu.Unlock() + + if len(state.TestResults) == 0 { + return 0 + } + + // 时间上限：interval=300s → 30min，interval=30s → 10min + // 防止 interval 过大时丢包率反映太久远的状态 + maxAge := 30 * time.Minute + interval := 300 + if interval > 0 { + age := time.Duration(6*interval) * time.Second + if age < 10*time.Minute { + age = 10 * time.Minute + } + if age > 30*time.Minute { + age = 30 * time.Minute + } + maxAge = age + } + + // 倒序遍历，最多取 20 个样本，超过 maxAge 的丢弃 + now := time.Now() + n, fails := 0, 0 + for i := len(state.TestResults) - 1; i >= 0 && n < 20; i-- { + if now.Sub(state.TestResults[i].At) > maxAge { + break + } + n++ + if !state.TestResults[i].Success { + fails++ + } + } + + // 样本不足 3 个不惩罚 + if n < 3 { + return 0 + } + // 置信度门控：3-5 个样本部分权重，5+ 个全权重 + confidence := float64(n) / 5.0 + if confidence > 1.0 { + confidence = 1.0 + } + return float64(fails) / float64(n) * confidence +} + +func (s *Smart) onDialFailed(proxy C.Proxy, err error) { + // 自修复组（URLTest/Fallback/LoadBalance/Smart）内部会切换 leaf，Smart 只需清除缓存重新评估 + // 非自修复组（Selector/Leaf）无自修复能力，Smart 需要主动标记退化并避开 + if isSelfHealingGroup(proxy.Type()) { + s.fastSingle.Reset() + } else { + s.markDegraded(proxy.Name()) + s.fastSingle.Reset() + } +} + +func (s *Smart) onDialSuccess() { + s.GroupBase.onDialSuccess() +} + +func isSelfHealingGroup(t C.AdapterType) bool { + return t == C.URLTest || t == C.Fallback || t == C.LoadBalance || t == C.Smart +} + +func (s *Smart) EffectiveSpeed() uint64 { + return s.fast(false).EffectiveSpeed() +} + +func percentile(values []float64, p float64) float64 { + if len(values) == 0 { + return 0 + } + sorted := make([]float64, len(values)) + copy(sorted, values) + for i := 0; i < len(sorted); i++ { + for j := i + 1; j < len(sorted); j++ { + if sorted[i] > sorted[j] { + sorted[i], sorted[j] = sorted[j], sorted[i] + } + } + } + index := p * float64(len(sorted)-1) + lower := int(math.Floor(index)) + upper := int(math.Ceil(index)) + if lower == upper { + return sorted[lower] + } + weight := index - float64(lower) + return sorted[lower]*(1-weight) + sorted[upper]*weight +} + +func medianUint16(values []uint16) uint64 { + if len(values) == 0 { + return 0 + } + sorted := make([]uint16, len(values)) + copy(sorted, values) + for i := 0; i < len(sorted); i++ { + for j := i + 1; j < len(sorted); j++ { + if sorted[i] > sorted[j] { + sorted[i], sorted[j] = sorted[j], sorted[i] + } + } + } + mid := len(sorted) / 2 + if len(sorted)%2 == 0 { + return (uint64(sorted[mid-1]) + uint64(sorted[mid])) / 2 + } + return uint64(sorted[mid]) +} + +func NewSmart(option *GroupCommonOption, providers []P.ProxyProvider, options ...smartOption) (*Smart, error) { + // 递归检查所有子节点（包括嵌套组内部），不允许 LoadBalance 作为子节点 + if err := checkNoLoadBalance(providers); err != nil { + return nil, err + } + + smart := &Smart{ + GroupBase: NewGroupBase(GroupBaseOption{ + Name: option.Name, + Type: C.Smart, + Hidden: option.Hidden, + Icon: option.Icon, + Filter: option.Filter, + ExcludeFilter: option.ExcludeFilter, + ExcludeType: option.ExcludeType, + TestTimeout: option.TestTimeout, + MaxFailedTimes: option.MaxFailedTimes, + Providers: providers, + }), + fastSingle: singledo.NewSingle[C.Proxy](10 * time.Second), + disableUDP: option.DisableUDP, + testUrl: option.URL, + expectedStatus: option.ExpectedStatus, + weightDelay: defaultWeightDelay, + weightLoss: defaultWeightLoss, + weightSpeed: defaultWeightSpeed, + tolerance: defaultTolerance, + } + + for _, opt := range options { + opt(smart) + } + + return smart, nil +} + +func checkNoLoadBalance(providers []P.ProxyProvider) error { + for _, pd := range providers { + for _, p := range pd.Proxies() { + if err := checkNoLoadBalanceRecursive(p); err != nil { + return err + } + } + } + return nil +} + +func checkNoLoadBalanceRecursive(p C.Proxy) error { + if p.Type() == C.LoadBalance { + return errors.New("smart group does not support LoadBalance as child: " + p.Name()) + } + if adapter := p.Adapter(); adapter != nil { + switch a := adapter.(type) { + case *URLTest: + for _, child := range a.GetProxies(false) { + if err := checkNoLoadBalanceRecursive(child); err != nil { + return err + } + } + case *Fallback: + for _, child := range a.GetProxies(false) { + if err := checkNoLoadBalanceRecursive(child); err != nil { + return err + } + } + case *Selector: + for _, child := range a.GetProxies(false) { + if err := checkNoLoadBalanceRecursive(child); err != nil { + return err + } + } + case *Smart: + for _, child := range a.GetProxies(false) { + if err := checkNoLoadBalanceRecursive(child); err != nil { + return err + } + } + } + } + return nil +} diff --git a/adapter/outboundgroup/urltest.go b/adapter/outboundgroup/urltest.go index 640c840c8e..8bbb9f672f 100644 --- a/adapter/outboundgroup/urltest.go +++ b/adapter/outboundgroup/urltest.go @@ -98,6 +98,11 @@ func (u *URLTest) Unwrap(metadata *C.Metadata, touch bool) C.Proxy { return u.fast(touch) } +// EffectiveSpeed implements C.Proxy - delegates to current leaf +func (u *URLTest) EffectiveSpeed() uint64 { + return u.fast(false).EffectiveSpeed() +} + func (u *URLTest) healthCheck() { u.fastSingle.Reset() u.GroupBase.healthCheck() diff --git a/constant/adapters.go b/constant/adapters.go index 3dcc715840..60841061a0 100644 --- a/constant/adapters.go +++ b/constant/adapters.go @@ -28,6 +28,7 @@ const ( Fallback URLTest LoadBalance + Smart Shadowsocks ShadowsocksR @@ -148,6 +149,11 @@ type DelayHistory struct { Delay uint16 `json:"delay"` } +type SpeedHistory struct { + Time time.Time `json:"time"` + Speed uint64 `json:"speed"` // bytes/sec +} + type ProxyState struct { Alive bool `json:"alive"` History []DelayHistory `json:"history"` @@ -163,6 +169,14 @@ type Proxy interface { ExtraDelayHistories() map[string]ProxyState LastDelayForTestUrl(url string) uint16 URLTest(ctx context.Context, url string, expectedStatus utils.IntRanges[uint16]) (uint16, error) + + SpeedHistory() []SpeedHistory + PushSpeed(speed uint64) + LastSpeed() uint64 + EffectiveSpeed() uint64 + + PacketLossRate(url string) float64 + PushTestResult(url string, success bool) } // AdapterType is enum of adapter type @@ -228,6 +242,8 @@ func (at AdapterType) String() string { return "URLTest" case LoadBalance: return "LoadBalance" + case Smart: + return "Smart" default: return "Unknown" } diff --git a/tunnel/statistic/manager.go b/tunnel/statistic/manager.go index c69746dbcb..af65ab6862 100644 --- a/tunnel/statistic/manager.go +++ b/tunnel/statistic/manager.go @@ -2,15 +2,23 @@ package statistic import ( "os" + syncatomic "sync/atomic" "time" "github.com/metacubex/mihomo/common/atomic" "github.com/metacubex/mihomo/common/xsync" "github.com/metacubex/mihomo/component/memory" + C "github.com/metacubex/mihomo/constant" ) var DefaultManager *Manager +var proxyResolver syncatomic.Value // func(string) C.Proxy + +func RegisterProxyResolver(fn func(string) C.Proxy) { + proxyResolver.Store(fn) +} + func init() { DefaultManager = &Manager{ uploadTemp: atomic.NewInt64(0), @@ -20,6 +28,7 @@ func init() { uploadTotal: atomic.NewInt64(0), downloadTotal: atomic.NewInt64(0), pid: int32(os.Getpid()), + speedObserver: NewSpeedObserver(), } go DefaultManager.handle() @@ -35,13 +44,16 @@ type Manager struct { downloadTotal atomic.Int64 pid int32 memory uint64 + speedObserver *SpeedObserver } func (m *Manager) Join(c Tracker) { m.connections.Store(c.ID(), c) + m.speedObserver.Join(c) } func (m *Manager) Leave(c Tracker) { + m.speedObserver.Leave(c) m.connections.Delete(c.ID()) } @@ -118,6 +130,10 @@ func (m *Manager) handle() { for range ticker.C { m.uploadBlip.Store(m.uploadTemp.Swap(0)) m.downloadBlip.Store(m.downloadTemp.Swap(0)) + + if resolve, ok := proxyResolver.Load().(func(string) C.Proxy); ok { + m.speedObserver.Tick(time.Now(), resolve) + } } } diff --git a/tunnel/statistic/speedobserver.go b/tunnel/statistic/speedobserver.go new file mode 100644 index 0000000000..6a43f95e7c --- /dev/null +++ b/tunnel/statistic/speedobserver.go @@ -0,0 +1,134 @@ +package statistic + +import ( + "sync" + "time" + + "github.com/metacubex/mihomo/common/atomic" + "github.com/metacubex/mihomo/common/xsync" + C "github.com/metacubex/mihomo/constant" +) + +const ( + minSpeedWindow = time.Second +) + +type trackerSample struct { + mu sync.Mutex + leaf string + lastTotal uint64 + windowBytes uint64 + windowStart time.Time + closed bool + tracker Tracker +} + +type SpeedObserver struct { + trackers xsync.Map[string, *trackerSample] + leaveTemp xsync.Map[string, *atomic.Uint64] +} + +func NewSpeedObserver() *SpeedObserver { + return &SpeedObserver{ + trackers: xsync.Map[string, *trackerSample]{}, + leaveTemp: xsync.Map[string, *atomic.Uint64]{}, + } +} + +func (o *SpeedObserver) Join(tt Tracker) { + info := tt.Info() + if info == nil || len(info.Chain) == 0 { + return + } + now := time.Now() + o.trackers.Store(tt.ID(), &trackerSample{ + leaf: info.Chain[0], + lastTotal: uint64(info.DownloadTotal.Load()), + windowStart: now, + tracker: tt, + }) +} + +func (o *SpeedObserver) Leave(tt Tracker) { + s, ok := o.trackers.LoadAndDelete(tt.ID()) + if !ok { + return + } + info := tt.Info() + if info == nil { + return + } + total := uint64(info.DownloadTotal.Load()) + if leaf, speed, emit := s.flush(time.Now(), total, true); emit { + slot, _ := o.leaveTemp.LoadOrStoreFn(leaf, func() *atomic.Uint64 { + return &atomic.Uint64{} + }) + slot.Add(speed) + } +} + +func (o *SpeedObserver) Tick(now time.Time, resolve func(string) C.Proxy) { + perProxy := make(map[string]uint64) + + o.trackers.Range(func(_ string, s *trackerSample) bool { + info := s.tracker.Info() + if info == nil { + return true + } + total := uint64(info.DownloadTotal.Load()) + if leaf, speed, emit := s.flush(now, total, false); emit { + perProxy[leaf] += speed + } + return true + }) + + o.leaveTemp.Range(func(leaf string, slot *atomic.Uint64) bool { + if v := slot.Swap(0); v > 0 { + perProxy[leaf] += v + } + return true + }) + + for leaf, speed := range perProxy { + if p := resolve(leaf); p != nil { + p.PushSpeed(speed) + } + } +} + +func (s *trackerSample) flush(now time.Time, total uint64, final bool) (string, uint64, bool) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.closed { + return "", 0, false + } + + delta := total - s.lastTotal + s.lastTotal = total + s.windowBytes += delta + + elapsed := now.Sub(s.windowStart) + if !final && elapsed < minSpeedWindow { + return "", 0, false + } + + if s.windowBytes == 0 { + if !final { + s.windowStart = now + } + s.closed = final + return "", 0, false + } + + denom := elapsed + if final && denom < minSpeedWindow { + denom = minSpeedWindow + } + + speed := uint64(float64(s.windowBytes) / denom.Seconds()) + s.windowBytes = 0 + s.windowStart = now + s.closed = final + return s.leaf, speed, true +} diff --git a/tunnel/tunnel.go b/tunnel/tunnel.go index fb2cc75b2e..ec1e44ac47 100644 --- a/tunnel/tunnel.go +++ b/tunnel/tunnel.go @@ -67,6 +67,12 @@ var ( ruleUpdateCallback = utils.NewCallback[P.RuleProvider]() ) +func init() { + statistic.RegisterProxyResolver(func(name string) C.Proxy { + return GetProxy(name) + }) +} + type tunnel struct{} var Tunnel = tunnel{} @@ -208,6 +214,12 @@ func Proxies() map[string]C.Proxy { return proxies } +func GetProxy(name string) C.Proxy { + configMux.RLock() + defer configMux.RUnlock() + return proxies[name] +} + // Providers return all compatible providers func Providers() map[string]P.ProxyProvider { return providers