优化自适应扫描系统 & 修复 POC 调度问题

自适应扫描优化: - target/ceiling 分离，自适应池可向上探索而非锁死在 target - assessHealth 阈值按网络环境区分（LAN 收紧 / Internet 放宽） - RTT 漂移时动态压低 target，配合 AIMD 双重降速 - 去掉 semaphore 双层流控，由 ants pool 统一反压 - 探测端口从 3 个扩充到 8 个，减少 RTT 采样偏差 - computeRetries 按环境调整目标概率和上限 Bug 修复: - AdaptivePool.Wait() 加 10 分钟超时，防止 goroutine 卡死时永久挂起 - CEL 环境初始化失败后允许重试（sync.Once → sync.Mutex + 标志位） - CAS 自旋加 runtime.Gosched() 退避，减少高并发下 CPU 空转 - -full 模式下 web 插件跳过 IsMarkedWebService 检查 #588 - 不确定服务补做 HTTP 回退探测，覆盖自定义框架漏网场景 - POC sets 纯字面量值跳过 CEL 编译，消除大量误报错误日志
2026-09-22 03:10:42 +08:00 · 2026-06-14 22:23:48 +08:00
parent c49c23c7f0
commit 8402be98e3
15 changed files with 798 additions and 104 deletions
@@ -0,0 +1,549 @@
+package core
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// =============================================================================
+// 优化 1：target/ceiling 分离
+// =============================================================================
+
+func TestOpt1_TargetCeilingSeparation_TuneConfig(t *testing.T) {
+	config := makeDefaultConfig()
+	session := makeTestSession(config)
+
+	ep := &EnvironmentProfile{
+		Net: NetworkProfile{
+			Env:       EnvLAN,
+			RTTMedian: 1 * time.Millisecond,
+			RTTStddev: 500 * time.Microsecond,
+			LossRate:  0.0,
+			Samples:   30,
+		},
+		System: SystemProfile{FDLimit: 65536, NumCPU: 8},
+	}
+
+	ep.TuneConfig(config, session)
+
+	if config.ThreadCeiling <= 0 {
+		t.Fatalf("ThreadCeiling 未被设置: %d", config.ThreadCeiling)
+	}
+
+	// 内网 factor=1.5，非显式 → target=ceiling=recommended
+	// 但 ceiling 应该 >= target
+	if config.ThreadCeiling < config.ThreadNum {
+		t.Errorf("Ceiling(%d) < ThreadNum(%d)", config.ThreadCeiling, config.ThreadNum)
+	}
+
+	t.Logf("target=%d, ceiling=%d", config.ThreadNum, config.ThreadCeiling)
+}
+
+func TestOpt1_TargetCeilingSeparation_ExplicitT(t *testing.T) {
+	config := makeDefaultConfig()
+	config.ThreadNum = 200
+	config.ThreadNumExplicit = true
+	session := makeTestSession(config)
+
+	ep := &EnvironmentProfile{
+		Net: NetworkProfile{
+			Env:       EnvInternet,
+			RTTMedian: 100 * time.Millisecond,
+			RTTStddev: 30 * time.Millisecond,
+			LossRate:  0.0,
+			Samples:   20,
+		},
+		System: SystemProfile{FDLimit: 65536, NumCPU: 8},
+	}
+
+	ep.TuneConfig(config, session)
+
+	// 用户显式指定 -t → ceiling = threadNum = 200
+	if config.ThreadCeiling != 200 {
+		t.Errorf("显式 -t 200: ceiling=%d, want 200", config.ThreadCeiling)
+	}
+	if config.ThreadNum != 200 {
+		t.Errorf("显式 -t 200: threadNum=%d, want 200", config.ThreadNum)
+	}
+}
+
+func TestOpt1_PoolUsesCeiling(t *testing.T) {
+	metrics := &ScanMetrics{}
+	target, ceiling := 50, 200
+
+	pool, err := NewAdaptivePool(target, ceiling, func(interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	pool.inSlowStart = false
+	pool.tune(target)
+
+	// 注入健康 metrics 让池增长
+	for i := 0; i < 200; i++ {
+		metrics.RecordConnect(time.Millisecond)
+	}
+
+	// 多次 adjust，池应能增长超过 target 但不超过 ceiling
+	for i := 0; i < 30; i++ {
+		pool.lastCheck.Store(0)
+		pool.adjust()
+	}
+
+	finalCap := pool.Cap()
+	if finalCap <= target {
+		t.Errorf("池应能超过 target(%d): cap=%d", target, finalCap)
+	}
+	if finalCap > ceiling {
+		t.Errorf("池不应超过 ceiling(%d): cap=%d", ceiling, finalCap)
+	}
+
+	t.Logf("target=%d, ceiling=%d, finalCap=%d", target, ceiling, finalCap)
+}
+
+func TestOpt1_FDLimitConstraintsBothFields(t *testing.T) {
+	config := makeDefaultConfig()
+	config.ThreadNum = 1000
+	session := makeTestSession(config)
+
+	ep := &EnvironmentProfile{
+		Net: NetworkProfile{
+			Env:       EnvLAN,
+			RTTMedian: 1 * time.Millisecond,
+			RTTStddev: 500 * time.Microsecond,
+			LossRate:  0.0,
+			Samples:   30,
+		},
+		System: SystemProfile{FDLimit: 256, NumCPU: 4},
+	}
+
+	ep.TuneConfig(config, session)
+
+	maxFD := 256 * 6 / 10
+	if config.ThreadNum > maxFD {
+		t.Errorf("ThreadNum(%d) 超过 fd 限制(%d)", config.ThreadNum, maxFD)
+	}
+	if config.ThreadCeiling > maxFD {
+		t.Errorf("ThreadCeiling(%d) 超过 fd 限制(%d)", config.ThreadCeiling, maxFD)
+	}
+}
+
+// =============================================================================
+// 优化 2：RTT 漂移微调 target
+// =============================================================================
+
+func TestOpt2_RTTDriftReducesTarget(t *testing.T) {
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(200, 400, func(interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	pool.inSlowStart = false
+	pool.tune(200)
+
+	// 建立基线：slow EMA 锚定在 1ms 附近
+	for i := 0; i < 500; i++ {
+		metrics.RecordConnect(1 * time.Millisecond)
+	}
+
+	origTarget := atomic.LoadInt32(&pool.target)
+
+	// RTT 突增到 100ms（100 倍），大量喂入让 fast EMA 拉开差距
+	for i := 0; i < 1000; i++ {
+		metrics.RecordConnect(100 * time.Millisecond)
+	}
+
+	ratio := metrics.RTTRatio()
+	t.Logf("RTT ratio after spike: %.2f", ratio)
+
+	if ratio <= 3.0 {
+		t.Skipf("RTT ratio=%.2f，EMA 差距不够大，跳过", ratio)
+	}
+
+	// 需要足够的新 metrics 让 assessHealth 的 deltaTotal >= 30
+	for i := 0; i < 50; i++ {
+		metrics.RecordConnect(100 * time.Millisecond)
+	}
+
+	// 多次 adjust 触发 maybeReduceTarget
+	for i := 0; i < 10; i++ {
+		pool.lastCheck.Store(0)
+		pool.prevSnapshot = MetricsSnapshot{} // 重置快照让 delta 足够
+		pool.adjust()
+	}
+
+	newTarget := atomic.LoadInt32(&pool.target)
+	if newTarget >= origTarget {
+		t.Errorf("RTT 漂移后 target 应降低: %d -> %d (ratio=%.2f)", origTarget, newTarget, ratio)
+	}
+
+	// 不应低于 ceiling/5
+	minTarget := atomic.LoadInt32(&pool.ceiling) / 5
+	if newTarget < minTarget {
+		t.Errorf("target(%d) 低于下限(%d)", newTarget, minTarget)
+	}
+
+	t.Logf("RTT drift: ratio=%.2f, target %d -> %d (min=%d)", ratio, origTarget, newTarget, minTarget)
+}
+
+func TestOpt2_NoReductionWhenStable(t *testing.T) {
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(200, 400, func(interface{}) {}, metrics)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	pool.inSlowStart = false
+	pool.tune(200)
+
+	// 稳定 RTT
+	for i := 0; i < 200; i++ {
+		metrics.RecordConnect(10 * time.Millisecond)
+	}
+
+	origTarget := atomic.LoadInt32(&pool.target)
+
+	for i := 0; i < 10; i++ {
+		pool.lastCheck.Store(0)
+		pool.adjust()
+	}
+
+	newTarget := atomic.LoadInt32(&pool.target)
+	if newTarget != origTarget {
+		t.Errorf("稳定 RTT 不应改变 target: %d -> %d", origTarget, newTarget)
+	}
+}
+
+// =============================================================================
+// 优化 3：assessHealth 阈值跟 NetworkEnv 关联
+// =============================================================================
+
+func TestOpt3_LANTighterThresholds(t *testing.T) {
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, EnvLAN)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	pool.inSlowStart = false
+	pool.tune(100)
+
+	// 10% exhaust rate — 对 LAN 来说应该是 Congested（阈值 8%）
+	for i := 0; i < 100; i++ {
+		if i < 10 {
+			metrics.RecordExhausted()
+		} else {
+			metrics.RecordConnect(time.Millisecond)
+		}
+	}
+
+	pool.lastCheck.Store(0)
+	pool.adjust()
+
+	if pool.Cap() >= 100 {
+		t.Errorf("LAN 10%% exhaust 应触发降速: cap=%d", pool.Cap())
+	}
+
+	t.Logf("LAN tight threshold: cap=%d (from 100)", pool.Cap())
+}
+
+func TestOpt3_InternetLooseThresholds(t *testing.T) {
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, EnvInternet)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	pool.inSlowStart = false
+	pool.tune(100)
+
+	// 10% exhaust rate — 对 Internet 来说不算 Congested（阈值 25%），应是 Stressed
+	for i := 0; i < 100; i++ {
+		if i < 10 {
+			metrics.RecordExhausted()
+		} else {
+			metrics.RecordConnect(10 * time.Millisecond)
+		}
+	}
+
+	pool.lastCheck.Store(0)
+	pool.adjust()
+	capAfter := pool.Cap()
+
+	// Internet 对 10% exhaust 只是 Stressed（×0.85），不是 Congested（×0.5）
+	if capAfter < 80 {
+		t.Errorf("Internet 10%% exhaust 不应大幅降速: cap=%d", capAfter)
+	}
+
+	t.Logf("Internet loose threshold: cap=%d (from 100)", capAfter)
+}
+
+func TestOpt3_EnvAffectsHealthDecision(t *testing.T) {
+	envs := []struct {
+		env  NetworkEnv
+		name string
+	}{
+		{EnvLAN, "LAN"},
+		{EnvWAN, "WAN"},
+		{EnvInternet, "Internet"},
+	}
+
+	var caps []int
+
+	for _, e := range envs {
+		metrics := &ScanMetrics{}
+		pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, e.env)
+		if err != nil {
+			t.Fatalf("创建池失败: %v", err)
+		}
+
+		pool.inSlowStart = false
+		pool.tune(100)
+
+		// 相同的 12% exhaust rate
+		for i := 0; i < 100; i++ {
+			if i < 12 {
+				metrics.RecordExhausted()
+			} else {
+				metrics.RecordConnect(time.Millisecond)
+			}
+		}
+
+		pool.lastCheck.Store(0)
+		pool.adjust()
+		caps = append(caps, pool.Cap())
+		pool.Release()
+
+		t.Logf("%s: cap=%d (12%% exhaust)", e.name, caps[len(caps)-1])
+	}
+
+	// LAN 反应最激烈（cap 最低），Internet 最宽容（cap 最高）
+	if caps[0] >= caps[2] {
+		t.Errorf("LAN cap(%d) 应 < Internet cap(%d) for same exhaust rate", caps[0], caps[2])
+	}
+}
+
+// =============================================================================
+// 优化 4：去掉 semaphore，ants 池天然反压
+// =============================================================================
+
+func TestOpt4_SemaphoreRemoved(t *testing.T) {
+	// 验证 portScanTask 结构体不再有 semaphore 字段
+	// 如果 semaphore 被加回来，这段代码编译就会报 "unknown field"
+	_ = portScanTask{
+		host: "127.0.0.1",
+		port: 80,
+		addr: "127.0.0.1:80",
+	}
+	t.Log("portScanTask 无 semaphore 字段，反压由 ants pool 统一管理")
+}
+
+// =============================================================================
+// 优化 5：扩充探测端口
+// =============================================================================
+
+func TestOpt5_ProbePortsExpanded(t *testing.T) {
+	if len(probePorts) < 5 {
+		t.Errorf("probePorts 只有 %d 个，应该扩充到至少 5 个", len(probePorts))
+	}
+
+	// 验证包含关键端口
+	required := map[int]bool{80: false, 443: false, 22: false}
+	for _, p := range probePorts {
+		if _, ok := required[p]; ok {
+			required[p] = true
+		}
+	}
+	for port, found := range required {
+		if !found {
+			t.Errorf("probePorts 缺少关键端口 %d", port)
+		}
+	}
+
+	// 验证没有重复
+	seen := make(map[int]bool)
+	for _, p := range probePorts {
+		if seen[p] {
+			t.Errorf("probePorts 有重复端口 %d", p)
+		}
+		seen[p] = true
+	}
+
+	t.Logf("probePorts = %v (%d 个)", probePorts, len(probePorts))
+}
+
+// =============================================================================
+// 优化 6：computeRetries 环境自适应
+// =============================================================================
+
+func TestOpt6_RetriesEnvAware(t *testing.T) {
+	lossRate := 0.3 // 30% 丢包
+
+	lanRetry := computeRetries(lossRate, EnvLAN)
+	wanRetry := computeRetries(lossRate, EnvWAN)
+	inetRetry := computeRetries(lossRate, EnvInternet)
+
+	// LAN 目标概率更严格(0.5%)，应该重试更多；但上限更低(4)
+	// Internet 目标概率更宽松(2%)，应该重试更少；但上限更高(6)
+	t.Logf("30%% loss: LAN=%d, WAN=%d, Internet=%d", lanRetry, wanRetry, inetRetry)
+
+	if lanRetry < 1 || lanRetry > 4 {
+		t.Errorf("LAN retry=%d, 应在 [1,4]", lanRetry)
+	}
+	if wanRetry < 1 || wanRetry > 5 {
+		t.Errorf("WAN retry=%d, 应在 [1,5]", wanRetry)
+	}
+	if inetRetry < 1 || inetRetry > 6 {
+		t.Errorf("Internet retry=%d, 应在 [1,6]", inetRetry)
+	}
+}
+
+func TestOpt6_RetriesMaxByEnv(t *testing.T) {
+	// 高丢包率，各环境应返回各自上限
+	lanMax := computeRetries(0.99, EnvLAN)
+	wanMax := computeRetries(0.99, EnvWAN)
+	inetMax := computeRetries(0.99, EnvInternet)
+
+	if lanMax != 4 {
+		t.Errorf("LAN max retry=%d, want 4", lanMax)
+	}
+	if wanMax != 5 {
+		t.Errorf("WAN max retry=%d, want 5", wanMax)
+	}
+	if inetMax != 6 {
+		t.Errorf("Internet max retry=%d, want 6", inetMax)
+	}
+}
+
+func TestOpt6_RetriesMathCorrectness(t *testing.T) {
+	envs := []struct {
+		env        NetworkEnv
+		targetProb float64
+		name       string
+	}{
+		{EnvLAN, 0.005, "LAN"},
+		{EnvWAN, 0.01, "WAN"},
+		{EnvInternet, 0.02, "Internet"},
+	}
+
+	for _, e := range envs {
+		for _, loss := range []float64{0.05, 0.10, 0.20, 0.30} {
+			retries := computeRetries(loss, e.env)
+			prob := 1.0
+			for i := 0; i < retries; i++ {
+				prob *= loss
+			}
+			// 重试后全失败概率应 < targetProb（除非被 clamp 了）
+			if prob >= e.targetProb && retries < 4 {
+				t.Errorf("%s loss=%.0f%% retries=%d: P=%.6f >= %.3f",
+					e.name, loss*100, retries, prob, e.targetProb)
+			}
+		}
+	}
+}
+
+// =============================================================================
+// 端到端集成：全链路验证
+// =============================================================================
+
+func TestOptAll_EndToEnd_LANToPool(t *testing.T) {
+	// 模拟内网探测 → TuneConfig → 创建池 → 池根据 env 自适应
+	profile := classifyNetwork(
+		makeDurations([]int{1, 1, 2, 2, 2, 3, 3, 3, 4, 5}),
+		0, 10,
+	)
+
+	config := makeDefaultConfig()
+	session := makeTestSession(config)
+	ep := &EnvironmentProfile{Net: *profile, System: SystemProfile{FDLimit: 65536, NumCPU: 8}}
+	ep.TuneConfig(config, session)
+
+	// 验证 env 被存储
+	if config.DetectedNetworkEnv != int(EnvLAN) {
+		t.Errorf("DetectedNetworkEnv=%d, want %d(LAN)", config.DetectedNetworkEnv, int(EnvLAN))
+	}
+
+	// 验证 ceiling 合理
+	if config.ThreadCeiling < config.ThreadNum {
+		t.Errorf("ceiling(%d) < target(%d)", config.ThreadCeiling, config.ThreadNum)
+	}
+
+	// 创建池并验证 env 传递
+	netEnv := NetworkEnv(config.DetectedNetworkEnv)
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(config.ThreadNum, config.ThreadCeiling, func(interface{}) {}, metrics, netEnv)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	if pool.networkEnv != EnvLAN {
+		t.Errorf("池的 networkEnv=%v, want LAN", pool.networkEnv)
+	}
+
+	t.Logf("端到端 LAN: target=%d ceiling=%d env=%v maxRetry=%d",
+		config.ThreadNum, config.ThreadCeiling, netEnv, config.MaxRetries)
+}
+
+func TestOptAll_EndToEnd_InternetToPool(t *testing.T) {
+	profile := classifyNetwork(
+		makeDurations([]int{60, 70, 80, 90, 100, 110, 120, 130, 140, 150}),
+		0, 10,
+	)
+
+	config := makeDefaultConfig()
+	session := makeTestSession(config)
+	ep := &EnvironmentProfile{Net: *profile, System: SystemProfile{FDLimit: 4096, NumCPU: 4}}
+	ep.TuneConfig(config, session)
+
+	if config.DetectedNetworkEnv != int(EnvInternet) {
+		t.Errorf("DetectedNetworkEnv=%d, want %d(Internet)", config.DetectedNetworkEnv, int(EnvInternet))
+	}
+
+	// 公网 target 应明显低于默认 600
+	if config.ThreadNum >= 600 {
+		t.Errorf("公网 threadNum=%d, 应 < 600", config.ThreadNum)
+	}
+
+	// ceiling 应 == target（非显式模式）
+	if config.ThreadCeiling != config.ThreadNum {
+		t.Errorf("非显式模式 ceiling(%d) != target(%d)", config.ThreadCeiling, config.ThreadNum)
+	}
+
+	netEnv := NetworkEnv(config.DetectedNetworkEnv)
+	metrics := &ScanMetrics{}
+	pool, err := NewAdaptivePool(config.ThreadNum, config.ThreadCeiling, func(interface{}) {}, metrics, netEnv)
+	if err != nil {
+		t.Fatalf("创建池失败: %v", err)
+	}
+	defer pool.Release()
+
+	// 注入 12% exhaust，Internet 环境应只是 Stressed 而不是 Congested
+	pool.inSlowStart = false
+	pool.tune(config.ThreadNum)
+	for i := 0; i < 100; i++ {
+		if i < 12 {
+			metrics.RecordExhausted()
+		} else {
+			metrics.RecordConnect(80 * time.Millisecond)
+		}
+	}
+	pool.lastCheck.Store(0)
+	pool.adjust()
+
+	// cap 不应被砍到一半以下（Stressed 只降 15%）
+	if pool.Cap() < config.ThreadNum*7/10 {
+		t.Errorf("Internet 12%% exhaust 降速过猛: %d -> %d", config.ThreadNum, pool.Cap())
+	}
+
+	t.Logf("端到端 Internet: target=%d ceiling=%d cap_after_stress=%d",
+		config.ThreadNum, config.ThreadCeiling, pool.Cap())
+}
+