feat: 自适应并发调度 — 网络探测 + AIMD + 参数智能推导

扫描前自动探测网络环境（RTT、丢包率、fd limit），基于探测数据推导 6 个关键参数，替代硬编码默认值： - Timeout: median_RTT + 4σ（覆盖 99.9% 正常连接） - ModuleThreadNum: target_concurrency / 30 - MaxRetries: ceil(log(0.01)/log(loss_rate))（全失败概率 <1%） - ICMPRate: 环境基准 × fd 系数 - PocNum: 跟随 ModuleThreadNum - DisablePing: 已有 ICMP 权限降级机制线程池从单信号（资源耗尽率）升级为 AIMD + 慢启动： - 慢启动：target/4 起步，500ms 翻倍 - 稳态 AIMD：健康 +5%，拥塞 ×0.5 - 双信号：资源耗尽率 + RTT 趋势（双 EMA）用户 -t 显式指定时作为 ceiling，探测仍调整其他参数。测试：单元 + 边界 + 集成 + 真实网络，core 包 580+ 用例全通过。
2026-09-25 20:51:52 +08:00 · 2026-06-12 09:46:02 +08:00
parent 683707fcd4
commit f883944b2b
21 changed files with 3287 additions and 251 deletions
@@ -1,7 +1,6 @@
 package core

 import (
-	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -11,139 +10,244 @@ import (
 	"github.com/shadow1ng/fscan/common/i18n"
 )

-// AdaptivePool 自适应线程池
-// 封装 ants.PoolWithFunc，支持根据资源耗尽率动态调整线程数
+// HealthSignal 健康评估结果
+type HealthSignal int
+
+const (
+	HealthUnknown   HealthSignal = iota // 样本不足，无法判断
+	HealthGood                          // 一切正常，可以提速
+	HealthOK                            // 正常，维持现状
+	HealthStressed                      // 有压力信号，轻微降速
+	HealthCongested                     // 明确拥塞，大幅降速
+)
+
+// AdaptivePool 自适应线程池（AIMD + 慢启动）
+//
+// 三阶段工作模式：
+//  1. 慢启动：从 target/4 起步，每个检查周期翻倍，直到达到 target 或检测到拥塞
+//  2. 稳态 AIMD：健康时加性增（+5% target），拥塞时乘性减（×0.5）
+//  3. 恢复上限受 ceiling 约束，不会无限增长
+//
+// 健康评估基于两个信号：
+//   - 资源耗尽率（fd/端口不足）
+//   - RTT 趋势（fast EMA / slow EMA）
 type AdaptivePool struct {
-	pool  *ants.PoolWithFunc
-	state *common.State
+	pool    *ants.PoolWithFunc
+	metrics *ScanMetrics

-	initialSize int
-	minSize     int
-	maxSize     int
-	currentSize int32 // 原子操作
+	// 并发控制
+	target      int32 // 探测推荐的目标值
+	ceiling     int32 // 绝对上限（用户指定或探测推荐）
+	currentSize int32

-	// 监控参数
-	checkInterval      time.Duration
-	lastCheckNano      atomic.Int64 // UnixNano
-	lastExhaustedCount int64
-	lastPacketCount    int64
+	// 慢启动
+	inSlowStart bool
+	ssThreshold int32 // 慢启动阈值（拥塞后降为当前值）

-	// 阈值
-	exhaustedThreshold float64 // 资源耗尽率阈值（触发降级）
-	recoveryThreshold  float64 // 恢复阈值（允许升级）
+	// 检查定时
+	checkInterval time.Duration
+	lastCheck     atomic.Int64 // UnixNano

-	mu sync.Mutex
+	// 增量计算
+	mu           sync.Mutex
+	prevSnapshot MetricsSnapshot
 }

 // NewAdaptivePool 创建自适应线程池
-func NewAdaptivePool(size int, fn func(interface{}), state *common.State) (*AdaptivePool, error) {
-	// 移除 WithPreAlloc(true)，在大规模扫描时预分配可能导致内存问题
-	pool, err := ants.NewPoolWithFunc(size, fn)
+// target: 目标并发数（来自 NetworkProfile.RecommendConcurrency）
+// ceiling: 最大并发上限
+// metrics: 共享的扫描度量（scanSinglePort 写入，pool 读取）
+func NewAdaptivePool(target, ceiling int, fn func(interface{}), metrics *ScanMetrics) (*AdaptivePool, error) {
+	// 慢启动初始值：target 的 25%，但不低于 10
+	initial := target / 4
+	if initial < 10 {
+		initial = 10
+	}
+	if initial > target {
+		initial = target
+	}
+
+	pool, err := ants.NewPoolWithFunc(initial, fn)
 	if err != nil {
 		return nil, err
 	}

-	minSize := size / 4
-	if minSize < 10 {
-		minSize = 10
-	}
-
 	return &AdaptivePool{
-		pool:               pool,
-		state:              state,
-		initialSize:        size,
-		minSize:            minSize,
-		maxSize:            size,
-		currentSize:        int32(size),
-		checkInterval:      time.Second,
-		exhaustedThreshold: 0.10, // 10% 资源耗尽率触发降级
-		recoveryThreshold:  0.02, // 2% 以下允许恢复
+		pool:          pool,
+		metrics:       metrics,
+		target:        int32(target),
+		ceiling:       int32(ceiling),
+		currentSize:   int32(initial),
+		inSlowStart:   true,
+		ssThreshold:   int32(target),
+		checkInterval: 500 * time.Millisecond,
 	}, nil
 }

-// Invoke 提交任务，并在适当时机检查是否需要调整线程数
+// Invoke 提交任务
 func (ap *AdaptivePool) Invoke(task interface{}) error {
 	ap.maybeAdjust()
 	return ap.pool.Invoke(task)
 }

-// maybeAdjust 检查并可能调整线程池大小
-// 使用原子 CAS 进行时间检查，99%+ 的调用零锁开销
+// maybeAdjust 周期性检查并调整并发数
 func (ap *AdaptivePool) maybeAdjust() {
-	lastCheck := ap.lastCheckNano.Load()
+	last := ap.lastCheck.Load()
 	now := time.Now().UnixNano()
-	if now-lastCheck < int64(ap.checkInterval) {
+	if now-last < int64(ap.checkInterval) {
 		return
 	}
-	if !ap.lastCheckNano.CompareAndSwap(lastCheck, now) {
-		return // 其他 goroutine 已在检查
-	}
-
-	// 获取当前计数
-	currentExhausted := ap.state.GetResourceExhaustedCount()
-	currentPackets := ap.state.GetPacketCount()
-
-	ap.mu.Lock()
-	// 计算增量（本周期内的耗尽率）
-	deltaExhausted := currentExhausted - ap.lastExhaustedCount
-	deltaPackets := currentPackets - ap.lastPacketCount
-
-	ap.lastExhaustedCount = currentExhausted
-	ap.lastPacketCount = currentPackets
-	ap.mu.Unlock()
-
-	// 需要足够的样本才能判断
-	if deltaPackets < 100 {
+	if !ap.lastCheck.CompareAndSwap(last, now) {
 		return
 	}

-	rate := float64(deltaExhausted) / float64(deltaPackets)
-	currentSize := int(atomic.LoadInt32(&ap.currentSize))
+	ap.adjust()
+}

-	if rate > ap.exhaustedThreshold && currentSize > ap.minSize {
-		// 降级：减少 20% 线程
-		newSize := int(float64(currentSize) * 0.8)
-		if newSize < ap.minSize {
-			newSize = ap.minSize
-		}
+func (ap *AdaptivePool) adjust() {
+	health := ap.assessHealth()
+	if health == HealthUnknown {
+		return
+	}
+
+	current := int(atomic.LoadInt32(&ap.currentSize))
+	target := int(atomic.LoadInt32(&ap.target))
+	ceiling := int(atomic.LoadInt32(&ap.ceiling))
+
+	var newSize int
+
+	if ap.inSlowStart {
+		newSize = ap.adjustSlowStart(health, current, target)
+	} else {
+		newSize = ap.adjustAIMD(health, current, target)
+	}
+
+	// 下限：ceiling 的 5%，但不低于 10
+	minSize := ceiling / 20
+	if minSize < 10 {
+		minSize = 10
+	}
+
+	if newSize < minSize {
+		newSize = minSize
+	}
+	if newSize > ceiling {
+		newSize = ceiling
+	}
+
+	if newSize != current {
 		ap.tune(newSize)
-		common.LogInfo(i18n.Tr("adaptive_pool_resource_exhausted", fmt.Sprintf("%.1f", rate*100), currentSize, newSize))
-	} else if rate < ap.recoveryThreshold && currentSize < ap.maxSize {
-		// 恢复：增加 10% 线程（保守恢复）
-		newSize := int(float64(currentSize) * 1.1)
-		if newSize > ap.maxSize {
-			newSize = ap.maxSize
+
+		// 显著变化时记录日志
+		delta := newSize - current
+		if delta < 0 {
+			delta = -delta
 		}
-		if newSize > currentSize {
-			ap.tune(newSize)
+		if delta > current/5 {
+			if newSize < current {
+				common.LogInfo(i18n.Tr("adaptive_pool_decrease", current, newSize))
+			} else {
+				common.LogDebug(i18n.Tr("adaptive_pool_increase", current, newSize))
+			}
 		}
 	}
 }

-// tune 调整线程池大小
+func (ap *AdaptivePool) adjustSlowStart(health HealthSignal, current, target int) int {
+	switch health {
+	case HealthCongested, HealthStressed:
+		// 退出慢启动，设置阈值
+		ap.ssThreshold = int32(current)
+		ap.inSlowStart = false
+		common.LogDebug(i18n.Tr("adaptive_pool_slowstart_exit", current))
+		return int(float64(current) * 0.5)
+	default:
+		// 翻倍
+		newSize := current * 2
+		if newSize >= target {
+			newSize = target
+			ap.inSlowStart = false
+		}
+		return newSize
+	}
+}
+
+func (ap *AdaptivePool) adjustAIMD(health HealthSignal, current, target int) int {
+	switch health {
+	case HealthCongested:
+		// 乘性减：×0.5
+		newSize := int(float64(current) * 0.5)
+		ap.ssThreshold = int32(newSize)
+		return newSize
+	case HealthStressed:
+		// 温和降低：×0.85
+		return int(float64(current) * 0.85)
+	case HealthGood:
+		// 加性增：+5% of target，至少 +1
+		inc := target / 20
+		if inc < 1 {
+			inc = 1
+		}
+		return current + inc
+	default:
+		return current
+	}
+}
+
+// assessHealth 综合健康评估
+func (ap *AdaptivePool) assessHealth() HealthSignal {
+	snap := ap.metrics.Snapshot()
+
+	ap.mu.Lock()
+	prev := ap.prevSnapshot
+	ap.prevSnapshot = snap
+	ap.mu.Unlock()
+
+	// 计算本周期增量
+	deltaTotal := snap.Total() - prev.Total()
+	deltaExhausted := snap.Exhausted - prev.Exhausted
+
+	// 样本不足
+	if deltaTotal < 30 {
+		return HealthUnknown
+	}
+
+	exhaustRate := float64(deltaExhausted) / float64(deltaTotal)
+	rttRatio := ap.metrics.RTTRatio()
+
+	// 多信号综合判断
+	switch {
+	case exhaustRate > 0.15:
+		return HealthCongested
+	case rttRatio > 2.5:
+		return HealthCongested
+	case exhaustRate > 0.05:
+		return HealthStressed
+	case rttRatio > 1.8:
+		return HealthStressed
+	case exhaustRate < 0.01 && rttRatio < 1.3:
+		return HealthGood
+	default:
+		return HealthOK
+	}
+}
+
 func (ap *AdaptivePool) tune(newSize int) {
 	ap.pool.Tune(newSize)
 	atomic.StoreInt32(&ap.currentSize, int32(newSize))
 }

 // Running 返回当前运行中的 goroutine 数量
-func (ap *AdaptivePool) Running() int {
-	return ap.pool.Running()
-}
+func (ap *AdaptivePool) Running() int { return ap.pool.Running() }

 // Cap 返回当前池容量
-func (ap *AdaptivePool) Cap() int {
-	return int(atomic.LoadInt32(&ap.currentSize))
-}
+func (ap *AdaptivePool) Cap() int { return int(atomic.LoadInt32(&ap.currentSize)) }

 // Release 释放线程池
-func (ap *AdaptivePool) Release() {
-	ap.pool.Release()
-}
+func (ap *AdaptivePool) Release() { ap.pool.Release() }

 // Wait 等待所有任务完成
 func (ap *AdaptivePool) Wait() {
-	// ants 没有原生 Wait，通过 Running() == 0 轮询
 	for ap.pool.Running() > 0 {
 		time.Sleep(10 * time.Millisecond)
 	}