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 04:31:52 +08:00 · 2026-06-14 22:23:45 +08:00
parent e0468ecd35
commit 88c7e4f2be
21 changed files with 3287 additions and 251 deletions
@@ -0,0 +1,545 @@
+package core
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// =============================================================================
+// 集成测试 1：探测 → 参数调整 → 线程池创建 完整链路
+// 验证从 NetworkProfile 到 TuneConfig 到 AdaptivePool 的端到端数据流
+// =============================================================================
+
+func TestIntegration_ProbeToPool_LAN(t *testing.T) {
+	// 模拟内网探测结果
+	profile := classifyNetwork(
+		makeDurations([]int{1, 1, 2, 2, 2, 3, 3, 3, 4, 5}), // ms
+		0, 10,
+	)
+
+	if profile.Env != EnvLAN {
+		t.Fatalf("探测环境 = %v, want LAN", profile.Env)
+	}
+
+	// 构建 Config + TuneConfig
+	config := makeDefaultConfig()
+	session := makeTestSession(config)
+	sys := ProbeSystem()
+
+	ep := &EnvironmentProfile{Net: *profile, System: sys}
+	ep.TuneConfig(config, session)
+
+	// 验证参数被合理调整
+	if config.Timeout > 3*time.Second {
+		t.Errorf("内网 Timeout = %v, 不应 > 3s", config.Timeout)
+	}
+	if config.MaxRetries != 1 {
+		t.Errorf("内网零丢包 MaxRetries = %d, want 1", config.MaxRetries)
+	}
+
+	// 用调整后的参数创建线程池
+	target, ceiling := profile.RecommendConcurrency(config.ThreadNum, config.ThreadNumExplicit)
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(target, ceiling, func(interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	if pool.Cap() <= 0 {
+		t.Errorf("池容量 = %d, 应该 > 0", pool.Cap())
+	}
+
+	t.Logf("内网完整链路: Timeout=%v MT=%d Retry=%d ICMP=%.2f target=%d ceiling=%d poolCap=%d",
+		config.Timeout, config.ModuleThreadNum, config.MaxRetries,
+		config.Network.ICMPRate, target, ceiling, pool.Cap())
+}
+
+func TestIntegration_ProbeToPool_Internet(t *testing.T) {
+	profile := classifyNetwork(
+		makeDurations([]int{60, 70, 80, 90, 100, 110, 120, 130, 140, 150}),
+		0, 10,
+	)
+
+	if profile.Env != EnvInternet {
+		t.Fatalf("探测环境 = %v, want Internet", profile.Env)
+	}
+
+	config := makeDefaultConfig()
+	session := makeTestSession(config)
+	ep := &EnvironmentProfile{Net: *profile, System: SystemProfile{FDLimit: 4096, NumCPU: 4}}
+	ep.TuneConfig(config, session)
+
+	target, ceiling := profile.RecommendConcurrency(config.ThreadNum, config.ThreadNumExplicit)
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(target, ceiling, func(interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	// 公网并发应该明显低于默认 600
+	if target >= 600 {
+		t.Errorf("公网 target = %d, 应该 < 600", target)
+	}
+
+	t.Logf("公网完整链路: Timeout=%v MT=%d Retry=%d target=%d ceiling=%d poolCap=%d",
+		config.Timeout, config.ModuleThreadNum, config.MaxRetries, target, ceiling, pool.Cap())
+}
+
+// =============================================================================
+// 集成测试 2：AdaptivePool + ScanMetrics 联动
+// 验证：任务执行 → metrics 记录 → 池读取 metrics → 做出调整决策
+// =============================================================================
+
+func TestIntegration_PoolMetrics_HealthyTraffic(t *testing.T) {
+	metrics := &ScanMetrics{}
+	var taskCount atomic.Int64
+
+	pool, err := NewAdaptivePool(100, 100, func(i interface{}) {
+		taskCount.Add(1)
+	}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	pool.inSlowStart = false
+	pool.tune(100)
+
+	// 注入健康 metrics
+	for i := 0; i < 200; i++ {
+		metrics.RecordConnect(time.Millisecond)
+	}
+
+	// 运行任务
+	var wg sync.WaitGroup
+	for i := 0; i < 200; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_ = pool.Invoke(nil)
+		}()
+	}
+	wg.Wait()
+	pool.Wait()
+
+	// 触发调整
+	pool.lastCheck.Store(0)
+	pool.adjust()
+
+	if pool.Cap() < 90 {
+		t.Errorf("健康流量池容量不应大幅下降: cap = %d", pool.Cap())
+	}
+
+	t.Logf("健康流量: tasks=%d connects=%d cap=%d",
+		taskCount.Load(), metrics.Snapshot().Connects, pool.Cap())
+}
+
+func TestIntegration_PoolMetrics_ExhaustedTraffic(t *testing.T) {
+	metrics := &ScanMetrics{}
+
+	pool, err := NewAdaptivePool(100, 100, func(i interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	pool.inSlowStart = false
+	pool.tune(100)
+
+	// 直接向 metrics 注入大量资源耗尽事件（模拟扫描过程中的 fd 不足）
+	for i := 0; i < 200; i++ {
+		metrics.RecordExhausted()
+	}
+
+	// 手动触发调整（清除时间守卫）
+	pool.lastCheck.Store(0)
+	pool.adjust()
+
+	// 资源耗尽率 100% → 应该降速
+	if pool.Cap() >= 100 {
+		t.Errorf("资源耗尽后池应该降速: cap = %d", pool.Cap())
+	}
+
+	t.Logf("资源耗尽: exhausted=%d cap=%d", metrics.Snapshot().Exhausted, pool.Cap())
+}
+
+// =============================================================================
+// 集成测试 3：慢启动 → 稳态 AIMD 过渡
+// 验证慢启动阶段的翻倍行为和过渡到稳态的时机
+// =============================================================================
+
+func TestIntegration_SlowStartToSteady(t *testing.T) {
+	metrics := &ScanMetrics{}
+
+	pool, err := NewAdaptivePool(100, 100, func(i interface{}) {
+		metrics.RecordConnect(time.Millisecond)
+	}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	if !pool.inSlowStart {
+		t.Fatal("初始应该在慢启动状态")
+	}
+
+	initialCap := pool.Cap()
+	t.Logf("慢启动初始: cap=%d", initialCap)
+
+	// 喂入足够的健康 metrics
+	for i := 0; i < 100; i++ {
+		metrics.RecordConnect(time.Millisecond)
+	}
+
+	// 模拟多次调整周期
+	caps := []int{initialCap}
+	for i := 0; i < 10; i++ {
+		pool.lastCheck.Store(0) // 强制触发检查
+		pool.adjust()
+		caps = append(caps, pool.Cap())
+	}
+
+	// 验证：容量应该逐步增长
+	growing := false
+	for i := 1; i < len(caps); i++ {
+		if caps[i] > caps[i-1] {
+			growing = true
+			break
+		}
+	}
+	if !growing {
+		t.Errorf("慢启动期间容量没有增长: %v", caps)
+	}
+
+	// 最终应该退出慢启动
+	finalCap := pool.Cap()
+	if finalCap < initialCap {
+		t.Errorf("最终容量 %d < 初始 %d, 不合理", finalCap, initialCap)
+	}
+
+	t.Logf("慢启动过渡: %v, inSlowStart=%v", caps, pool.inSlowStart)
+}
+
+// =============================================================================
+// 集成测试 4：拥塞 → 降速 → 恢复 完整周期
+// =============================================================================
+
+func TestIntegration_CongestionRecovery(t *testing.T) {
+	metrics := &ScanMetrics{}
+
+	pool, err := NewAdaptivePool(200, 200, func(i interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	// 直接到稳态，满容量
+	pool.inSlowStart = false
+	pool.tune(200)
+
+	// === 阶段 1: 正常运行 ===
+	for i := 0; i < 100; i++ {
+		metrics.RecordConnect(time.Millisecond)
+	}
+	pool.lastCheck.Store(0)
+	pool.adjust()
+	normalCap := pool.Cap()
+	t.Logf("正常阶段: cap=%d", normalCap)
+
+	// === 阶段 2: 突发拥塞（大量资源耗尽）===
+	for i := 0; i < 200; i++ {
+		metrics.RecordExhausted()
+	}
+	pool.lastCheck.Store(0)
+	pool.adjust()
+	congestedCap := pool.Cap()
+
+	if congestedCap >= normalCap {
+		t.Errorf("拥塞后应降速: normal=%d congested=%d", normalCap, congestedCap)
+	}
+	t.Logf("拥塞阶段: cap=%d (降幅 %d%%)", congestedCap, (normalCap-congestedCap)*100/normalCap)
+
+	// === 阶段 3: 恢复（大量成功连接）===
+	for i := 0; i < 500; i++ {
+		metrics.RecordConnect(time.Millisecond)
+	}
+
+	// 多次调整模拟恢复过程
+	for i := 0; i < 20; i++ {
+		pool.lastCheck.Store(0)
+		pool.adjust()
+	}
+	recoveredCap := pool.Cap()
+
+	if recoveredCap <= congestedCap {
+		t.Errorf("恢复后应提速: congested=%d recovered=%d", congestedCap, recoveredCap)
+	}
+
+	// 恢复后不应超过 ceiling
+	if recoveredCap > 200 {
+		t.Errorf("恢复后不应超过 ceiling: cap=%d ceiling=200", recoveredCap)
+	}
+
+	t.Logf("恢复阶段: cap=%d", recoveredCap)
+}
+
+// =============================================================================
+// 集成测试 5：RTT 趋势检测 → 池调整
+// 验证 ScanMetrics 的 RTT EMA 趋势信号能正确传导到池的健康判断
+// =============================================================================
+
+func TestIntegration_RTTTrend_DrivesPoolAdjustment(t *testing.T) {
+	metrics := &ScanMetrics{}
+
+	pool, err := NewAdaptivePool(100, 100, func(i interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	pool.inSlowStart = false
+	pool.tune(100)
+
+	// 建立基线：100 个 5ms RTT
+	for i := 0; i < 200; i++ {
+		metrics.RecordConnect(5 * time.Millisecond)
+	}
+	pool.lastCheck.Store(0)
+	pool.adjust()
+	baselineCap := pool.Cap()
+
+	// RTT 突增到 100ms（20 倍）
+	for i := 0; i < 100; i++ {
+		metrics.RecordConnect(100 * time.Millisecond)
+	}
+
+	ratio := metrics.RTTRatio()
+	if ratio <= 1.0 {
+		t.Logf("RTT ratio = %.2f, EMA 可能还没追上（正常）", ratio)
+	}
+
+	// 多次调整看池是否响应
+	for i := 0; i < 5; i++ {
+		pool.lastCheck.Store(0)
+		pool.adjust()
+	}
+	afterRTTSpike := pool.Cap()
+
+	t.Logf("RTT 趋势: baseline_cap=%d after_spike=%d rtt_ratio=%.2f",
+		baselineCap, afterRTTSpike, ratio)
+
+	// 如果 ratio 足够高，池应该降速
+	if ratio > 2.0 && afterRTTSpike >= baselineCap {
+		t.Errorf("RTT ratio=%.2f 但池没有降速: %d -> %d", ratio, baselineCap, afterRTTSpike)
+	}
+}
+
+// =============================================================================
+// 集成测试 6：不同网络环境下的参数一致性
+// 验证同一组目标在不同环境下参数调整的合理递进关系
+// =============================================================================
+
+func TestIntegration_ParameterProgression(t *testing.T) {
+	environments := []struct {
+		name    string
+		rtts    []int // ms
+		loss    int   // failures out of 10
+		wantEnv NetworkEnv
+	}{
+		{"内网", []int{1, 1, 2, 2, 3, 3, 4, 4, 5, 5}, 0, EnvLAN},
+		{"局域网", []int{10, 15, 20, 25, 30, 35, 40, 45, 48, 49}, 0, EnvWAN},
+		{"公网", []int{60, 70, 80, 90, 100, 120, 140, 160, 180, 195}, 0, EnvInternet},
+		{"慢速", []int{200, 300, 400, 500, 600, 700, 800, 900, 1000, 1500}, 0, EnvSlow},
+	}
+
+	type params struct {
+		timeout  time.Duration
+		mt       int
+		retry    int
+		icmpRate float64
+	}
+
+	var results []params
+
+	for _, env := range environments {
+		profile := classifyNetwork(makeDurations(env.rtts), env.loss, 10)
+		if profile.Env != env.wantEnv {
+			t.Errorf("%s: env = %v, want %v", env.name, profile.Env, env.wantEnv)
+		}
+
+		config := makeDefaultConfig()
+		session := makeTestSession(config)
+		ep := &EnvironmentProfile{
+			Net:    *profile,
+			System: SystemProfile{FDLimit: 65536, NumCPU: 8},
+		}
+		ep.TuneConfig(config, session)
+
+		results = append(results, params{
+			timeout:  config.Timeout,
+			mt:       config.ModuleThreadNum,
+			retry:    config.MaxRetries,
+			icmpRate: config.Network.ICMPRate,
+		})
+
+		t.Logf("%s: Timeout=%v MT=%d Retry=%d ICMP=%.2f",
+			env.name, config.Timeout, config.ModuleThreadNum, config.MaxRetries, config.Network.ICMPRate)
+	}
+
+	// 验证递进关系：从内网到慢速，Timeout 应递增
+	for i := 1; i < len(results); i++ {
+		if results[i].timeout < results[i-1].timeout {
+			t.Errorf("Timeout 不递增: %v (env[%d]) < %v (env[%d])",
+				results[i].timeout, i, results[i-1].timeout, i-1)
+		}
+	}
+
+	// ICMPRate 应递减（内网最高，慢速最低）
+	for i := 1; i < len(results); i++ {
+		if results[i].icmpRate > results[i-1].icmpRate {
+			t.Errorf("ICMPRate 不递减: %.2f (env[%d]) > %.2f (env[%d])",
+				results[i].icmpRate, i, results[i-1].icmpRate, i-1)
+		}
+	}
+}
+
+// =============================================================================
+// 集成测试 7：用户显式 -t + 网络探测 完整流程
+// 验证用户指定值作为 ceiling 但探测仍然影响其他参数
+// =============================================================================
+
+func TestIntegration_ExplicitThreadNum_WithProbe(t *testing.T) {
+	profile := classifyNetwork(
+		makeDurations([]int{100, 120, 140, 160, 180, 200, 220, 240, 260, 300}),
+		2, 12, // 部分丢包
+	)
+
+	config := makeDefaultConfig()
+	config.ThreadNum = 200
+	config.ThreadNumExplicit = true
+	session := makeTestSession(config)
+
+	ep := &EnvironmentProfile{
+		Net:    *profile,
+		System: SystemProfile{FDLimit: 4096, NumCPU: 4},
+	}
+	ep.TuneConfig(config, session)
+
+	// ThreadNum 不应被修改（fd limit 允许范围内）
+	// 但 Timeout、ModuleThreadNum 等应根据探测调整
+	if config.Timeout == 3*time.Second {
+		t.Error("即使 -t 显式，Timeout 仍应根据探测调整")
+	}
+
+	// 创建池
+	target, ceiling := profile.RecommendConcurrency(config.ThreadNum, config.ThreadNumExplicit)
+	if ceiling != 200 {
+		t.Errorf("显式 -t 200 的 ceiling = %d, want 200", ceiling)
+	}
+	if target > 200 {
+		t.Errorf("target = %d, 不应超过 ceiling 200", target)
+	}
+
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(target, ceiling, func(interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	t.Logf("显式 -t 200: Timeout=%v MT=%d Retry=%d target=%d ceiling=%d cap=%d",
+		config.Timeout, config.ModuleThreadNum, config.MaxRetries, target, ceiling, pool.Cap())
+}
+
+// =============================================================================
+// 集成测试 8：AdaptiveTimeout + ScanMetrics 双 RTT 追踪
+// 验证两个 RTT 追踪器独立工作不干扰
+// =============================================================================
+
+func TestIntegration_DualRTTTracking(t *testing.T) {
+	adaptiveTO := NewAdaptiveTimeout(3 * time.Second)
+	metrics := &ScanMetrics{}
+
+	// 喂入相同的 RTT 数据到两个追踪器
+	for i := 0; i < 50; i++ {
+		rtt := 10 * time.Millisecond
+		adaptiveTO.Record(rtt)
+		metrics.RecordConnect(rtt)
+	}
+
+	// AdaptiveTimeout 用于连接超时
+	toValue := adaptiveTO.Timeout()
+	// ScanMetrics 用于池健康判断
+	rttFast := metrics.RTTFast()
+	ratio := metrics.RTTRatio()
+
+	if toValue > 3*time.Second {
+		t.Errorf("AdaptiveTimeout 应该 < 初始值: %v", toValue)
+	}
+	if rttFast < 8*time.Millisecond || rttFast > 12*time.Millisecond {
+		t.Errorf("ScanMetrics RTTFast 应接近 10ms: %v", rttFast)
+	}
+	if ratio < 0.8 || ratio > 1.2 {
+		t.Errorf("稳定 RTT 的 ratio 应接近 1.0: %.2f", ratio)
+	}
+
+	t.Logf("双追踪: AdaptiveTO=%v, MetricsFast=%v, Ratio=%.2f", toValue, rttFast, ratio)
+}
+
+// =============================================================================
+// 集成测试 9：丢包环境下 Retry + ModuleThreadNum 联动
+// 验证高丢包同时影响重试和并发
+// =============================================================================
+
+func TestIntegration_LossyNetwork_RetryAndConcurrency(t *testing.T) {
+	lossRates := []float64{0.0, 0.05, 0.10, 0.20, 0.40}
+
+	type result struct {
+		loss  float64
+		retry int
+		mt    int
+	}
+	var results []result
+
+	for _, loss := range lossRates {
+		profile := &NetworkProfile{
+			Env:       EnvInternet,
+			RTTMedian: 80 * time.Millisecond,
+			RTTStddev: 20 * time.Millisecond,
+			LossRate:  loss,
+			Samples:   20,
+		}
+
+		config := makeDefaultConfig()
+		session := makeTestSession(config)
+		ep := &EnvironmentProfile{
+			Net:    *profile,
+			System: SystemProfile{FDLimit: 65536, NumCPU: 8},
+		}
+		ep.TuneConfig(config, session)
+
+		results = append(results, result{loss, config.MaxRetries, config.ModuleThreadNum})
+	}
+
+	// 重试次数应随丢包率单调递增
+	for i := 1; i < len(results); i++ {
+		if results[i].retry < results[i-1].retry {
+			t.Errorf("Retry 不递增: loss=%.2f retry=%d < loss=%.2f retry=%d",
+				results[i].loss, results[i].retry, results[i-1].loss, results[i-1].retry)
+		}
+	}
+
+	// 高丢包时 ModuleThreadNum 应降低
+	if results[len(results)-1].mt >= results[0].mt {
+		t.Errorf("40%%丢包的 MT(%d) 应 < 0%%丢包的 MT(%d)",
+			results[len(results)-1].mt, results[0].mt)
+	}
+
+	for _, r := range results {
+		t.Logf("loss=%.0f%%: Retry=%d MT=%d", r.loss*100, r.retry, r.mt)
+	}
+}