From f883944b2bb51591d2ed0738ef7c32419c4d0245 Mon Sep 17 00:00:00 2001 From: ZacharyZcR Date: Fri, 12 Jun 2026 09:46:02 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=87=AA=E9=80=82=E5=BA=94=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E8=B0=83=E5=BA=A6=20=E2=80=94=20=E7=BD=91=E7=BB=9C?= =?UTF-8?q?=E6=8E=A2=E6=B5=8B=20+=20AIMD=20+=20=E5=8F=82=E6=95=B0=E6=99=BA?= =?UTF-8?q?=E8=83=BD=E6=8E=A8=E5=AF=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 扫描前自动探测网络环境（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+ 用例全通过。 --- common/config_struct.go | 13 +- common/flag.go | 7 + common/flag_config.go | 14 +- common/i18n/locales/en.yaml | 20 ++ common/i18n/locales/zh.yaml | 20 ++ core/adaptive_pool.go | 276 ++++++++++----- core/adaptive_pool_test.go | 203 ++++------- core/edge_cases_test.go | 639 ++++++++++++++++++++++++++++++++++ core/env_profiler.go | 221 ++++++++++++ core/env_profiler_test.go | 334 ++++++++++++++++++ core/fd_limit_unix.go | 13 + core/fd_limit_windows.go | 8 + core/integration_test.go | 545 +++++++++++++++++++++++++++++ core/network_profiler.go | 277 +++++++++++++++ core/network_profiler_test.go | 169 +++++++++ core/port_scan.go | 30 +- core/real_network_test.go | 487 ++++++++++++++++++++++++++ core/scan_metrics.go | 115 ++++++ core/scan_metrics_test.go | 130 +++++++ core/service_scanner.go | 12 + web/api/scan.go | 5 + 21 files changed, 3287 insertions(+), 251 deletions(-) create mode 100644 core/edge_cases_test.go create mode 100644 core/env_profiler.go create mode 100644 core/env_profiler_test.go create mode 100644 core/fd_limit_unix.go create mode 100644 core/fd_limit_windows.go create mode 100644 core/integration_test.go create mode 100644 core/network_profiler.go create mode 100644 core/network_profiler_test.go create mode 100644 core/real_network_test.go create mode 100644 core/scan_metrics.go create mode 100644 core/scan_metrics_test.go diff --git a/common/config_struct.go b/common/config_struct.go index 9be2534..3a6e9ed 100644 --- a/common/config_struct.go +++ b/common/config_struct.go @@ -22,12 +22,13 @@ config_struct.go - 配置结构体定义 // Config 扫描器完整配置 - 初始化后只读，可安全共享 type Config struct { // 高频访问字段 - 平铺到顶层 - Timeout time.Duration // 通用超时 - ThreadNum int // 主线程数 - ModuleThreadNum int // 模块线程数 - DisableBrute bool // 禁用暴力破解 - DisablePing bool // 禁用Ping检测 - DisableTcpProbe bool // 禁用TCP补充探测 + Timeout time.Duration // 通用超时 + ThreadNum int // 主线程数 + ThreadNumExplicit bool // 用户显式指定了 -t + ModuleThreadNum int // 模块线程数 + DisableBrute bool // 禁用暴力破解 + DisablePing bool // 禁用Ping检测 + DisableTcpProbe bool // 禁用TCP补充探测 // 扫描模式 Mode string // 扫描模式 diff --git a/common/flag.go b/common/flag.go index 2844ba9..7a7c82a 100644 --- a/common/flag.go +++ b/common/flag.go @@ -213,6 +213,13 @@ func Flag(Info *HostInfo) error { return err } + // 检测用户是否显式指定了 -t + flag.Visit(func(f *flag.Flag) { + if f.Name == "t" { + fv.ThreadNumExplicit = true + } + }) + // 设置语言 i18n.SetLanguage(fv.Language) diff --git a/common/flag_config.go b/common/flag_config.go index cca579b..b147227 100644 --- a/common/flag_config.go +++ b/common/flag_config.go @@ -30,9 +30,10 @@ type FlagVars struct { PortsFile string // 扫描控制 - ScanMode string - ThreadNum int - ModuleThreadNum int + ScanMode string + ThreadNum int + ThreadNumExplicit bool // 用户显式指定了 -t + ModuleThreadNum int TimeoutSec int64 // 秒，需转换为 time.Duration GlobalTimeout int64 DisablePing bool @@ -134,9 +135,10 @@ func GetFlagVars() *FlagVars { func BuildConfigFromFlags(fv *FlagVars) *Config { return &Config{ // 高频字段 - Timeout: time.Duration(fv.TimeoutSec) * time.Second, - ThreadNum: fv.ThreadNum, - ModuleThreadNum: fv.ModuleThreadNum, + Timeout: time.Duration(fv.TimeoutSec) * time.Second, + ThreadNum: fv.ThreadNum, + ThreadNumExplicit: fv.ThreadNumExplicit, + ModuleThreadNum: fv.ModuleThreadNum, DisableBrute: fv.DisableBrute, DisablePing: fv.DisablePing, DisableTcpProbe: fv.DisableTcpProbe, diff --git a/common/i18n/locales/en.yaml b/common/i18n/locales/en.yaml index 3b480f4..e9a4007 100644 --- a/common/i18n/locales/en.yaml +++ b/common/i18n/locales/en.yaml @@ -541,6 +541,26 @@ icmp_debug_stable_done: other: "[ICMP] response stable, ending early, elapsed {{.Arg1}}, alive {{.Arg2}}/{{.Arg3}}" adaptive_pool_resource_exhausted: other: "[AdaptivePool] resource exhaustion rate {{.Arg1}}%, threads {{.Arg2}} -> {{.Arg3}}" +adaptive_pool_decrease: + other: "Concurrency adjusted: {{.Arg1}} -> {{.Arg2}} (pressure detected)" +adaptive_pool_increase: + other: "Concurrency adjusted: {{.Arg1}} -> {{.Arg2}} (network healthy)" +adaptive_pool_slowstart_exit: + other: "Slow start exit: current {{.Arg1}} (congestion detected)" +net_probe_result: + other: "Network probe: {{.Arg1}}, RTT {{.Arg2}}ms, loss {{.Arg3}}%, concurrency {{.Arg4}}/{{.Arg5}}" +net_env_lan: + other: "LAN" +net_env_wan: + other: "WAN" +net_env_internet: + other: "Internet" +net_env_slow: + other: "Slow network" +env_tune_summary: + other: "Adaptive params: Timeout={{.Arg1}}ms, ModuleThread={{.Arg2}}, Retry={{.Arg3}}, ICMPRate={{.Arg4}}, PocNum={{.Arg5}}" +env_fd_limit: + other: "fd limit constraint: threads {{.Arg1}} -> {{.Arg2}} (ulimit={{.Arg3}})" # ========================= Service Plugin Messages ========================= # Format: {service}_{type} - type: credential/unauth/service/vuln diff --git a/common/i18n/locales/zh.yaml b/common/i18n/locales/zh.yaml index 0659d02..4fd75a2 100644 --- a/common/i18n/locales/zh.yaml +++ b/common/i18n/locales/zh.yaml @@ -541,6 +541,26 @@ icmp_debug_stable_done: other: "[ICMP] 响应稳定，提前结束，耗时 {{.Arg1}}，存活 {{.Arg2}}/{{.Arg3}}" adaptive_pool_resource_exhausted: other: "[AdaptivePool] 资源耗尽率 {{.Arg1}}%, 线程数 {{.Arg2}} -> {{.Arg3}}" +adaptive_pool_decrease: + other: "并发调整: {{.Arg1}} -> {{.Arg2}} (检测到压力)" +adaptive_pool_increase: + other: "并发调整: {{.Arg1}} -> {{.Arg2}} (网络健康)" +adaptive_pool_slowstart_exit: + other: "慢启动退出: 当前 {{.Arg1}} (检测到拥塞)" +net_probe_result: + other: "网络探测: {{.Arg1}}, RTT {{.Arg2}}ms, 丢包 {{.Arg3}}%, 并发 {{.Arg4}}/{{.Arg5}}" +net_env_lan: + other: "内网" +net_env_wan: + other: "局域网" +net_env_internet: + other: "公网" +net_env_slow: + other: "慢速网络" +env_tune_summary: + other: "参数自适应: Timeout={{.Arg1}}ms, ModuleThread={{.Arg2}}, Retry={{.Arg3}}, ICMPRate={{.Arg4}}, PocNum={{.Arg5}}" +env_fd_limit: + other: "fd limit 约束: 线程数 {{.Arg1}} -> {{.Arg2}} (ulimit={{.Arg3}})" # ========================= 服务插件通用消息 ========================= # 格式: {service}_{type} - type: credential/unauth/service/vuln diff --git a/core/adaptive_pool.go b/core/adaptive_pool.go index d07041e..a4a7933 100644 --- a/core/adaptive_pool.go +++ b/core/adaptive_pool.go @@ -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) } diff --git a/core/adaptive_pool_test.go b/core/adaptive_pool_test.go index 392d362..5873399 100644 --- a/core/adaptive_pool_test.go +++ b/core/adaptive_pool_test.go @@ -1,160 +1,101 @@ package core -/* -adaptive_pool_test.go - AdaptivePool 高价值测试 - -测试重点： -1. 并发安全 - 多goroutine同时调整不崩溃 -2. 降级逻辑 - 资源耗尽率高时正确减少线程 -3. 恢复逻辑 - 资源耗尽率低时正确增加线程 -4. 边界条件 - 不超过minSize/maxSize - -不测试： -- 简单的getter方法（太简单，不值得） -- ants库本身的正确性（库作者负责） -*/ - import ( "testing" "time" - - "github.com/shadow1ng/fscan/common" ) -// ============================================================================= -// 场景1：降级逻辑测试（高价值） -// ============================================================================= - -// TestAdaptivePool_DowngradeOnHighExhaustion 验证资源耗尽率高时降低线程数 -// 这是个核心业务逻辑：耗尽率 > 10% 时应该减少线程 -func TestAdaptivePool_DowngradeOnHighExhaustion(t *testing.T) { - state := common.NewState() - - pool, err := NewAdaptivePool(100, func(interface{}) {}, state) +// newTestPool 测试辅助：创建测试用的自适应线程池 +func newTestPool(t *testing.T, size int, fn func(interface{})) (*AdaptivePool, *ScanMetrics) { + t.Helper() + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(size, size, fn, metrics) if err != nil { t.Fatalf("创建线程池失败: %v", err) } + return pool, metrics +} + +// TestAdaptivePool_DowngradeOnHighExhaustion 验证资源耗尽率高时降低线程数 +func TestAdaptivePool_DowngradeOnHighExhaustion(t *testing.T) { + pool, metrics := newTestPool(t, 100, func(interface{}) {}) defer pool.Release() + // 慢启动先跑到 target + pool.inSlowStart = false + pool.tune(100) + initialCap := pool.Cap() - // 模拟高资源耗尽率：20% 的包都失败了 - // 需要至少100个样本才会触发调整 + // 模拟高资源耗尽率：20% for i := 0; i < 200; i++ { - state.IncrementPacketCount() - if i < 40 { // 前40个失败（20%） - state.IncrementResourceExhaustedCount() + if i < 40 { + metrics.RecordExhausted() + } else { + metrics.RecordConnect(time.Millisecond) } } - // 触发调整：提交足够多的任务让maybeAdjust被调用 + // 触发调整 for i := 0; i < 20; i++ { _ = pool.Invoke(nil) - time.Sleep(time.Millisecond * 10) // 等待异步调整 + time.Sleep(time.Millisecond * 30) } - // 等待调整完成 - time.Sleep(time.Millisecond * 50) - finalCap := pool.Cap() - // 验证：线程数应该减少 if finalCap >= initialCap { t.Errorf("应该降级: 初始 %d, 最终 %d", initialCap, finalCap) } - // 验证：不应该降到minSize以下 - minSize := initialCap / 4 - if minSize < 10 { - minSize = 10 - } - if finalCap < minSize { - t.Errorf("降到minSize以下: %d < %d", finalCap, minSize) + if finalCap < 10 { + t.Errorf("降到 minSize 以下: %d", finalCap) } - t.Logf("降级成功: %d -> %d (min=%d)", initialCap, finalCap, minSize) + t.Logf("降级成功: %d -> %d", initialCap, finalCap) } -// ============================================================================= -// 场景3：恢复逻辑测试（高价值） -// ============================================================================= - -// TestAdaptivePool_NoRecoveryOnLowExhaustion 验证低耗尽率时不升级 -// 防止线程数盲目增长 -func TestAdaptivePool_NoRecoveryOnLowExhaustion(t *testing.T) { - state := common.NewState() - - pool, err := NewAdaptivePool(50, func(interface{}) {}, state) +// TestAdaptivePool_SlowStart 验证慢启动行为 +func TestAdaptivePool_SlowStart(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics) if err != nil { t.Fatalf("创建线程池失败: %v", err) } defer pool.Release() - // 先降到minSize - for i := 0; i < 500; i++ { - state.IncrementPacketCount() - state.IncrementResourceExhaustedCount() // 100% 耗尽 + // 初始应该是 target/4 = 25 + initialCap := pool.Cap() + if initialCap > 30 { + t.Errorf("慢启动初始值应该 <= 30, got %d", initialCap) } - for i := 0; i < 20; i++ { - _ = pool.Invoke(nil) - } - time.Sleep(time.Millisecond * 50) - - reducedCap := pool.Cap() - - // 现在模拟低耗尽率：只有1%失败 - for i := 0; i < 500; i++ { - state.IncrementPacketCount() - if i%100 == 0 { // 只有5个失败（1%） - state.IncrementResourceExhaustedCount() - } + if !pool.inSlowStart { + t.Error("应该处于慢启动状态") } - for i := 0; i < 20; i++ { - _ = pool.Invoke(nil) - } - time.Sleep(time.Millisecond * 50) - - finalCap := pool.Cap() - - // 验证：即使耗尽率低，也不应该立即恢复（保守策略） - // 或者即使恢复，也很有限 - if finalCap > reducedCap+5 { - t.Logf("恢复行为: %d -> %d", reducedCap, finalCap) - } + t.Logf("慢启动初始: cap=%d, inSlowStart=%v", initialCap, pool.inSlowStart) } -// ============================================================================= -// 场景4：边界条件测试（中价值） -// ============================================================================= - -// TestAdaptivePool_MinSizeBoundary 验证不会降到minSize以下 +// TestAdaptivePool_MinSizeBoundary 验证不会降到 minSize 以下 func TestAdaptivePool_MinSizeBoundary(t *testing.T) { - state := common.NewState() - - // 创建小线程池，minSize会是10 - pool, err := NewAdaptivePool(40, func(interface{}) {}, state) - if err != nil { - t.Fatalf("创建线程池失败: %v", err) - } + pool, metrics := newTestPool(t, 40, func(interface{}) {}) defer pool.Release() - // 模拟极端的资源耗尽：100%失败 - for i := 0; i < 1000; i++ { - state.IncrementPacketCount() - state.IncrementResourceExhaustedCount() + pool.inSlowStart = false + pool.tune(40) + + // 极端耗尽 + for i := 0; i < 500; i++ { + metrics.RecordExhausted() } - // 触发多次调整 for i := 0; i < 50; i++ { _ = pool.Invoke(nil) - time.Sleep(time.Millisecond) + time.Sleep(time.Millisecond * 15) } finalCap := pool.Cap() - - // 验证：不应该低于10 if finalCap < 10 { t.Errorf("线程数 < 10: %d", finalCap) } @@ -162,76 +103,52 @@ func TestAdaptivePool_MinSizeBoundary(t *testing.T) { t.Logf("最小边界测试通过: cap=%d", finalCap) } -// ============================================================================= -// 场景5：样本不足测试（低价值但重要） -// ============================================================================= - // TestAdaptivePool_NotEnoughSamples 验证样本不足时不调整 -// 防止基于小样本做错误决策 func TestAdaptivePool_NotEnoughSamples(t *testing.T) { - state := common.NewState() - - pool, err := NewAdaptivePool(100, func(interface{}) {}, state) - if err != nil { - t.Fatalf("创建线程池失败: %v", err) - } + pool, metrics := newTestPool(t, 100, func(interface{}) {}) defer pool.Release() + pool.inSlowStart = false + pool.tune(100) initialCap := pool.Cap() - // 只增加少量样本（<100），不足以触发调整 - for i := 0; i < 50; i++ { - state.IncrementPacketCount() - state.IncrementResourceExhaustedCount() // 即使100%失败也不调整 + // 只 20 个样本，不足 30 的阈值 + for i := 0; i < 20; i++ { + metrics.RecordExhausted() } - // 提交任务 for i := 0; i < 10; i++ { _ = pool.Invoke(nil) } time.Sleep(time.Millisecond * 50) finalCap := pool.Cap() - - // 验证：样本不足时不应该调整 if finalCap != initialCap { t.Errorf("样本不足时不应该调整: %d -> %d", initialCap, finalCap) } } -// ============================================================================= -// 辅助函数 -// ============================================================================= - -// TestAdaptivePool_Wait 验证Wait方法正确等待所有任务完成 +// TestAdaptivePool_Wait 验证 Wait 方法 func TestAdaptivePool_Wait(t *testing.T) { - state := common.NewState() - - pool, err := NewAdaptivePool(10, func(interface{}) { + pool, _ := newTestPool(t, 10, func(interface{}) { time.Sleep(time.Millisecond * 50) - }, state) - if err != nil { - t.Fatalf("创建线程池失败: %v", err) - } + }) defer pool.Release() - // 提交任务 + pool.inSlowStart = false + pool.tune(10) + for i := 0; i < 20; i++ { _ = pool.Invoke(nil) } - // Wait应该在所有任务完成后返回 start := time.Now() pool.Wait() duration := time.Since(start) - // 20个任务，每个50ms，10个线程，应该约100ms完成 - if duration < 80*time.Millisecond { - t.Logf("Wait提前返回？可能测试有问题: %v", duration) - } - if duration > 200*time.Millisecond { - t.Errorf("Wait耗时过长: %v", duration) + if duration > 300*time.Millisecond { + t.Errorf("Wait 耗时过长: %v", duration) } - t.Logf("Wait测试通过: %v", duration) + t.Logf("Wait 测试通过: %v", duration) } diff --git a/core/edge_cases_test.go b/core/edge_cases_test.go new file mode 100644 index 0000000..854235c --- /dev/null +++ b/core/edge_cases_test.go @@ -0,0 +1,639 @@ +package core + +import ( + "math" + "sync" + "testing" + "time" +) + +// ============================================================================= +// computeRetries 边界 +// ============================================================================= + +func TestComputeRetries_EdgeCases(t *testing.T) { + tests := []struct { + lossRate float64 + wantMin int + wantMax int + desc string + }{ + {-0.5, 1, 1, "负数丢包率: 视为零"}, + {-1.0, 1, 1, "负一: 视为零"}, + {0.0, 1, 1, "精确零"}, + {0.001, 1, 1, "精确边界 0.001"}, + {0.0009, 1, 1, "低于 0.001 边界"}, + {0.0011, 1, 6, "高于 0.001 边界"}, + {0.95, 6, 6, "精确边界 0.95"}, + {0.949, 1, 6, "低于 0.95 边界"}, + {0.951, 6, 6, "高于 0.95 边界"}, + {1.0, 6, 6, "精确 1.0"}, + {1.5, 6, 6, "超过 1.0"}, + {100.0, 6, 6, "极大值"}, + {math.SmallestNonzeroFloat64, 1, 1, "最小正浮点数"}, + } + + for _, tt := range tests { + t.Run(tt.desc, func(t *testing.T) { + got := computeRetries(tt.lossRate) + if got < tt.wantMin || got > tt.wantMax { + t.Errorf("computeRetries(%v) = %d, want [%d, %d]", + tt.lossRate, got, tt.wantMin, tt.wantMax) + } + if got < 1 || got > 6 { + t.Errorf("computeRetries(%v) = %d, 超出 [1,6] 范围", tt.lossRate, got) + } + }) + } +} + +func TestComputeRetries_NaN_Inf(t *testing.T) { + // 确保不 panic + for _, v := range []float64{math.NaN(), math.Inf(1), math.Inf(-1)} { + got := computeRetries(v) + if got < 1 || got > 6 { + t.Errorf("computeRetries(%v) = %d, 超出 [1,6] 范围", v, got) + } + } +} + +// ============================================================================= +// computeICMPRate 边界 +// ============================================================================= + +func TestComputeICMPRate_EdgeCases(t *testing.T) { + tests := []struct { + env NetworkEnv + fdLimit int + desc string + }{ + {EnvLAN, 1, "fd=1: 极小"}, + {EnvLAN, -1, "fd=负数: 应被忽略"}, + {EnvLAN, 0, "fd=0: 未知"}, + {EnvLAN, math.MaxInt32, "fd=极大"}, + {NetworkEnv(99), 1024, "未知环境类型"}, + } + + for _, tt := range tests { + t.Run(tt.desc, func(t *testing.T) { + net := &NetworkProfile{Env: tt.env} + sys := &SystemProfile{FDLimit: tt.fdLimit} + got := computeICMPRate(net, sys) + if got <= 0 || math.IsNaN(got) || math.IsInf(got, 0) { + t.Errorf("computeICMPRate(env=%v, fd=%d) = %v, 无效值", tt.env, tt.fdLimit, got) + } + }) + } +} + +// ============================================================================= +// classifyEnv 精确边界值 +// ============================================================================= + +func TestClassifyEnv_ExactBoundaries(t *testing.T) { + tests := []struct { + median time.Duration + lossRate float64 + want NetworkEnv + desc string + }{ + // RTT 边界 + {4999 * time.Microsecond, 0.0, EnvLAN, "4.999ms → LAN"}, + {5 * time.Millisecond, 0.0, EnvWAN, "精确 5ms → WAN"}, + {49999 * time.Microsecond, 0.0, EnvWAN, "49.999ms → WAN"}, + {50 * time.Millisecond, 0.0, EnvInternet, "精确 50ms → Internet"}, + {199999 * time.Microsecond, 0.0, EnvInternet, "199.999ms → Internet"}, + {200 * time.Millisecond, 0.0, EnvSlow, "精确 200ms → Slow"}, + + // 丢包率边界 + {1 * time.Millisecond, 0.009, EnvLAN, "丢包 0.9% → LAN"}, + {1 * time.Millisecond, 0.01, EnvWAN, "精确 1% → WAN (不满足 < 0.01)"}, + {1 * time.Millisecond, 0.011, EnvWAN, "丢包 1.1% → WAN (超过 LAN 阈值)"}, + {20 * time.Millisecond, 0.049, EnvWAN, "丢包 4.9% → WAN"}, + {20 * time.Millisecond, 0.05, EnvInternet, "精确 5% → Internet (不满足 < 0.05)"}, + {20 * time.Millisecond, 0.051, EnvInternet, "丢包 5.1% → Internet"}, + {1 * time.Millisecond, 0.099, EnvInternet, "丢包 9.9% → Internet"}, + {1 * time.Millisecond, 0.10, EnvInternet, "精确 10% → Internet (< 判断)"}, + {1 * time.Millisecond, 0.101, EnvSlow, "丢包 10.1% → Slow"}, + + // 零值 + {0, 0.0, EnvLAN, "零 RTT 零丢包 → LAN"}, + } + + for _, tt := range tests { + t.Run(tt.desc, func(t *testing.T) { + got := classifyEnv(tt.median, tt.lossRate) + if got != tt.want { + t.Errorf("classifyEnv(median=%v, loss=%.4f) = %v, want %v", + tt.median, tt.lossRate, got, tt.want) + } + }) + } +} + +// ============================================================================= +// classifyNetwork 边界 +// ============================================================================= + +func TestClassifyNetwork_EdgeCases(t *testing.T) { + t.Run("单个 RTT 样本", func(t *testing.T) { + p := classifyNetwork([]time.Duration{5 * time.Millisecond}, 0, 1) + if p.Samples != 1 { + t.Errorf("samples = %d, want 1", p.Samples) + } + // stddev 应该是 0 + if p.RTTStddev != 0 { + t.Errorf("单样本 stddev = %v, want 0", p.RTTStddev) + } + }) + + t.Run("所有 RTT 相同", func(t *testing.T) { + rtts := make([]time.Duration, 50) + for i := range rtts { + rtts[i] = 10 * time.Millisecond + } + p := classifyNetwork(rtts, 0, 50) + if p.RTTStddev != 0 { + t.Errorf("全相同 RTT stddev = %v, want 0", p.RTTStddev) + } + if p.RTTMedian != 10*time.Millisecond { + t.Errorf("median = %v, want 10ms", p.RTTMedian) + } + }) + + t.Run("极大 RTT 值", func(t *testing.T) { + rtts := []time.Duration{time.Hour, time.Hour, time.Hour} + p := classifyNetwork(rtts, 0, 3) + if p.Env != EnvSlow { + t.Errorf("env = %v, want Slow", p.Env) + } + }) + + t.Run("混合极端值", func(t *testing.T) { + rtts := []time.Duration{time.Microsecond, time.Hour} + p := classifyNetwork(rtts, 0, 2) + // 不 panic 就行 + if p.Samples != 2 { + t.Errorf("samples = %d, want 2", p.Samples) + } + }) + + t.Run("全部失败无响应", func(t *testing.T) { + p := classifyNetwork(nil, 100, 100) + if p.Env != EnvWAN { + t.Errorf("env = %v, want WAN (default)", p.Env) + } + }) + + t.Run("failures > total (异常输入)", func(t *testing.T) { + rtts := []time.Duration{time.Millisecond} + p := classifyNetwork(rtts, 10, 5) // failures > total + // lossRate = 1 - 1/5 = 0.8, 不应 panic + if p.LossRate < 0 { + t.Errorf("lossRate = %.2f, 不应为负", p.LossRate) + } + }) + + t.Run("total=0", func(t *testing.T) { + p := classifyNetwork(nil, 0, 0) + // 不 panic + if p.Samples != 0 { + t.Errorf("samples = %d, want 0", p.Samples) + } + }) +} + +// ============================================================================= +// RecommendConcurrency 边界 +// ============================================================================= + +func TestRecommendConcurrency_EdgeCases(t *testing.T) { + tests := []struct { + env NetworkEnv + loss float64 + userT int + explicit bool + desc string + }{ + {EnvLAN, 0.0, 0, false, "userThreadNum=0"}, + {EnvLAN, 0.0, 1, false, "userThreadNum=1"}, + {EnvLAN, 0.0, -1, false, "userThreadNum 负数"}, + {EnvLAN, 0.0, math.MaxInt32, false, "userThreadNum 极大"}, + {EnvLAN, 0.99, 600, false, "99% 丢包"}, + {EnvLAN, 1.0, 600, false, "100% 丢包"}, + {EnvSlow, 0.0, 1, true, "慢速+显式+1"}, + } + + for _, tt := range tests { + t.Run(tt.desc, func(t *testing.T) { + p := &NetworkProfile{Env: tt.env, LossRate: tt.loss, Samples: 10} + target, ceiling := p.RecommendConcurrency(tt.userT, tt.explicit) + // 不 panic，且 target >= 1（clamp 保底 10 或 userT） + if target < 0 || ceiling < 0 { + t.Errorf("target=%d ceiling=%d, 不应为负", target, ceiling) + } + if tt.explicit && ceiling != tt.userT && tt.userT > 0 { + t.Errorf("显式模式 ceiling=%d, want %d", ceiling, tt.userT) + } + t.Logf("env=%v loss=%.2f userT=%d explicit=%v → target=%d ceiling=%d", + tt.env, tt.loss, tt.userT, tt.explicit, target, ceiling) + }) + } +} + +// ============================================================================= +// ScanMetrics 边界 +// ============================================================================= + +func TestScanMetrics_EdgeCases(t *testing.T) { + t.Run("RTT=0", func(t *testing.T) { + m := &ScanMetrics{} + m.RecordConnect(0) + // 不 panic + if m.Total() != 1 { + t.Errorf("Total = %d, want 1", m.Total()) + } + }) + + t.Run("负数 RTT", func(t *testing.T) { + m := &ScanMetrics{} + m.RecordConnect(-time.Millisecond) + // 不 panic，负数 RTT 应被忽略 + if m.rttSamples.Load() != 0 { + t.Errorf("负数 RTT 不应计入采样: got %d", m.rttSamples.Load()) + } + }) + + t.Run("极大 RTT", func(t *testing.T) { + m := &ScanMetrics{} + m.RecordConnect(time.Hour) + if m.RTTFast() != time.Hour { + t.Errorf("首个样本 RTTFast = %v, want 1h", m.RTTFast()) + } + }) + + t.Run("EMA 首个样本初始化", func(t *testing.T) { + m := &ScanMetrics{} + m.RecordConnect(10 * time.Millisecond) + if m.rttFastNs.Load() != int64(10*time.Millisecond) { + t.Errorf("首个样本应直接设置 EMA: got %d", m.rttFastNs.Load()) + } + }) + + t.Run("空 Snapshot", func(t *testing.T) { + m := &ScanMetrics{} + snap := m.Snapshot() + if snap.Total() != 0 { + t.Errorf("空 metrics Snapshot.Total = %d, want 0", snap.Total()) + } + }) + + t.Run("RTTRatio 单侧为零", func(t *testing.T) { + m := &ScanMetrics{} + // 手动设置一个但不设另一个——不应该发生，但防御 + m.rttFastNs.Store(1000) + m.rttSlowNs.Store(0) + m.rttSamples.Store(30) + ratio := m.RTTRatio() + if ratio != 1.0 { + t.Errorf("slow=0 时 ratio = %.2f, want 1.0", ratio) + } + }) + + t.Run("大量操作不溢出", func(t *testing.T) { + m := &ScanMetrics{} + for i := 0; i < 100000; i++ { + m.RecordConnect(time.Millisecond) + } + if m.Total() != 100000 { + t.Errorf("Total = %d, want 100000", m.Total()) + } + ratio := m.RTTRatio() + if math.IsNaN(ratio) || math.IsInf(ratio, 0) { + t.Errorf("大量样本后 ratio = %v, 不应为 NaN/Inf", ratio) + } + }) +} + +// ============================================================================= +// TuneConfig 边界 +// ============================================================================= + +func TestTuneConfig_EdgeCases(t *testing.T) { + t.Run("RTTMedian=0 RTTStddev=0", func(t *testing.T) { + config := makeDefaultConfig() + session := makeTestSession(config) + ep := &EnvironmentProfile{ + Net: NetworkProfile{Env: EnvLAN, RTTMedian: 0, RTTStddev: 0, Samples: 10}, + System: SystemProfile{FDLimit: 65536}, + } + ep.TuneConfig(config, session) + // Timeout: median(0) + 4*stddev(0) = 0 → minTO = 0+200ms → clamp to 1s + if config.Timeout < time.Second { + t.Errorf("零 RTT Timeout = %v, 应该 >= 1s", config.Timeout) + } + }) + + t.Run("RTTStddev 远大于 RTTMedian", func(t *testing.T) { + config := makeDefaultConfig() + session := makeTestSession(config) + ep := &EnvironmentProfile{ + Net: NetworkProfile{Env: EnvInternet, RTTMedian: 10 * time.Millisecond, RTTStddev: 5 * time.Second, Samples: 10}, + System: SystemProfile{FDLimit: 65536}, + } + ep.TuneConfig(config, session) + // Timeout = 10ms + 4*5s = 20.01s → clamp to 10s + if config.Timeout != 10*time.Second { + t.Errorf("极大 stddev Timeout = %v, 应该被 clamp 到 10s", config.Timeout) + } + }) + + t.Run("ThreadNum=0", func(t *testing.T) { + config := makeDefaultConfig() + config.ThreadNum = 0 + session := makeTestSession(config) + ep := &EnvironmentProfile{ + Net: NetworkProfile{Env: EnvLAN, RTTMedian: time.Millisecond, RTTStddev: time.Millisecond, Samples: 10}, + System: SystemProfile{FDLimit: 65536}, + } + ep.TuneConfig(config, session) + // ModuleThreadNum = 0/30 = 0 → clamp to 5 + if config.ModuleThreadNum < 5 { + t.Errorf("ThreadNum=0 时 ModuleThreadNum = %d, 应该 >= 5", config.ModuleThreadNum) + } + }) + + t.Run("多次调用 TuneConfig", func(t *testing.T) { + config := makeDefaultConfig() + session := makeTestSession(config) + ep := &EnvironmentProfile{ + Net: NetworkProfile{Env: EnvLAN, RTTMedian: time.Millisecond, RTTStddev: time.Millisecond, LossRate: 0.0, Samples: 10}, + System: SystemProfile{FDLimit: 65536}, + } + + ep.TuneConfig(config, session) + first := config.Timeout + + // 第二次调用——已经调整过的值不等于默认值，应被视为"显式" + ep.TuneConfig(config, session) + second := config.Timeout + + if first != second { + t.Errorf("多次调用 TuneConfig 不应重复调整: %v vs %v", first, second) + } + }) + + t.Run("fd limit = ThreadNum 精确值", func(t *testing.T) { + config := makeDefaultConfig() + config.ThreadNum = 600 + session := makeTestSession(config) + ep := &EnvironmentProfile{ + Net: NetworkProfile{Samples: 0}, + System: SystemProfile{FDLimit: 1000}, // 1000 * 0.6 = 600 + } + ep.TuneConfig(config, session) + // ThreadNum(600) == maxConcurrency(600), 不应触发约束 + if config.ThreadNum != 600 { + t.Errorf("fd=1000 时 ThreadNum = %d, 不应被约束", config.ThreadNum) + } + }) + + t.Run("fd limit 精确低于 ThreadNum", func(t *testing.T) { + config := makeDefaultConfig() + config.ThreadNum = 600 + session := makeTestSession(config) + ep := &EnvironmentProfile{ + Net: NetworkProfile{Samples: 0}, + System: SystemProfile{FDLimit: 999}, // 999 * 0.6 = 599 + } + ep.TuneConfig(config, session) + if config.ThreadNum > 599 { + t.Errorf("fd=999 时 ThreadNum = %d, 应该 <= 599", config.ThreadNum) + } + }) +} + +// ============================================================================= +// AdaptivePool 边界 +// ============================================================================= + +func TestAdaptivePool_EdgeCases(t *testing.T) { + t.Run("target=1", func(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(1, 1, func(interface{}) {}, metrics) + if err != nil { + t.Fatalf("创建失败: %v", err) + } + defer pool.Release() + // initial = max(1/4, 10) = 10 → 但 10 > target(1)... 看实现 + // 实际上 initial = min(max(1/4, 10), 1) = 1... 不对 + // initial = target/4 = 0, 但 < 10, 所以 initial = 10 + // 但 initial > target(1)... initial = min(10, 1) = 1 + // 看代码：if initial > target { initial = target } + if pool.Cap() != 1 { + t.Errorf("target=1 时 cap = %d, want 1", pool.Cap()) + } + }) + + t.Run("target=0", func(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(0, 0, func(interface{}) {}, metrics) + // ants 可能拒绝 size=0 + if err != nil { + t.Logf("target=0 正确返回错误: %v", err) + return + } + defer pool.Release() + t.Logf("target=0 cap = %d", pool.Cap()) + }) + + t.Run("ceiling < target", func(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(100, 50, func(interface{}) {}, metrics) + if err != nil { + t.Fatalf("创建失败: %v", err) + } + defer pool.Release() + // initial = 100/4 = 25, 不超过 ceiling + if pool.Cap() > 50 { + t.Errorf("ceiling=50 但 cap = %d", pool.Cap()) + } + }) + + t.Run("高频 Invoke 不 panic", func(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(10, 10, func(interface{}) { + time.Sleep(time.Millisecond) + }, metrics) + if err != nil { + t.Fatalf("创建失败: %v", err) + } + defer pool.Release() + pool.inSlowStart = false + pool.tune(10) + + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _ = pool.Invoke(nil) + }() + } + wg.Wait() + pool.Wait() + }) + + t.Run("assessHealth 零增量", func(t *testing.T) { + metrics := &ScanMetrics{} + pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics) + if err != nil { + t.Fatalf("创建失败: %v", err) + } + defer pool.Release() + + // 初始化 prevSnapshot 后不产生新数据 + pool.prevSnapshot = metrics.Snapshot() + health := pool.assessHealth() + if health != HealthUnknown { + t.Errorf("零增量应返回 HealthUnknown, got %v", health) + } + }) +} + +// ============================================================================= +// pickSamples 边界 +// ============================================================================= + +func TestPickSamples_EdgeCases(t *testing.T) { + t.Run("maxSamples=0", func(t *testing.T) { + s := pickSamples([]string{"a", "b"}, 0) + if len(s) != 0 { + t.Errorf("maxSamples=0 应返回空, got %d", len(s)) + } + }) + + t.Run("maxSamples=1", func(t *testing.T) { + s := pickSamples([]string{"a", "b", "c"}, 1) + if len(s) != 1 { + t.Errorf("maxSamples=1 应返回 1 个, got %d", len(s)) + } + }) + + t.Run("hosts 等于 maxSamples", func(t *testing.T) { + hosts := []string{"a", "b", "c"} + s := pickSamples(hosts, 3) + if len(s) != 3 { + t.Errorf("应返回全部, got %d", len(s)) + } + }) +} + +// ============================================================================= +// isTimeoutError / isConnectionRefused 边界 +// ============================================================================= + +func TestIsTimeoutError_EdgeCases(t *testing.T) { + if isTimeoutError(nil) { + t.Error("nil 不应判为 timeout") + } +} + +func TestIsConnectionRefused_EdgeCases(t *testing.T) { + if isConnectionRefused(nil) { + t.Error("nil 不应判为 refused") + } +} + +// ============================================================================= +// NetworkEnv.String 覆盖 +// ============================================================================= + +func TestNetworkEnv_String(t *testing.T) { + for _, env := range []NetworkEnv{EnvLAN, EnvWAN, EnvInternet, EnvSlow} { + s := env.String() + if s == "" { + t.Errorf("NetworkEnv(%d).String() = 空", env) + } + } + // 未知值 + s := NetworkEnv(99).String() + if s == "" { + t.Error("未知 NetworkEnv.String() = 空") + } +} + +// ============================================================================= +// clampInt / clampDuration 边界 +// ============================================================================= + +func TestClampInt(t *testing.T) { + tests := []struct { + v, min, max, want int + }{ + {5, 1, 10, 5}, + {0, 1, 10, 1}, + {15, 1, 10, 10}, + {-5, -10, -1, -5}, + {5, 5, 5, 5}, // min == max == v + {3, 5, 5, 5}, // v < min == max + {10, 5, 5, 5}, // v > min == max + } + + for _, tt := range tests { + got := clampInt(tt.v, tt.min, tt.max) + if got != tt.want { + t.Errorf("clampInt(%d, %d, %d) = %d, want %d", tt.v, tt.min, tt.max, got, tt.want) + } + } +} + +func TestClampDuration(t *testing.T) { + got := clampDuration(5*time.Second, time.Second, 10*time.Second) + if got != 5*time.Second { + t.Errorf("got %v, want 5s", got) + } + got = clampDuration(0, time.Second, 10*time.Second) + if got != time.Second { + t.Errorf("got %v, want 1s", got) + } + got = clampDuration(time.Hour, time.Second, 10*time.Second) + if got != 10*time.Second { + t.Errorf("got %v, want 10s", got) + } +} + +// ============================================================================= +// isExplicit 边界 +// ============================================================================= + +func TestIsExplicit(t *testing.T) { + config := makeDefaultConfig() + // 默认值 → 非显式 + if isExplicit(config, "time") { + t.Error("默认 Timeout 不应视为显式") + } + if isExplicit(config, "mt") { + t.Error("默认 ModuleThreadNum 不应视为显式") + } + if isExplicit(config, "retry") { + t.Error("默认 MaxRetries 不应视为显式") + } + if isExplicit(config, "icmp-rate") { + t.Error("默认 ICMPRate 不应视为显式") + } + if isExplicit(config, "num") { + t.Error("默认 PocNum 不应视为显式") + } + + // 未知 flag + if isExplicit(config, "nonexistent") { + t.Error("未知 flag 不应视为显式") + } + + // ThreadNumExplicit + config.ThreadNumExplicit = true + if !isExplicit(config, "t") { + t.Error("ThreadNumExplicit=true 应视为显式") + } +} diff --git a/core/env_profiler.go b/core/env_profiler.go new file mode 100644 index 0000000..dde2267 --- /dev/null +++ b/core/env_profiler.go @@ -0,0 +1,221 @@ +package core + +import ( + "fmt" + "math" + "runtime" + "time" + + "github.com/shadow1ng/fscan/common" + "github.com/shadow1ng/fscan/common/i18n" +) + +// EnvironmentProfile 综合环境探测结果 +type EnvironmentProfile struct { + Net NetworkProfile + System SystemProfile +} + +// SystemProfile 系统能力信息 +type SystemProfile struct { + FDLimit int // 文件描述符上限（0 表示未知） + NumCPU int +} + +// ProbeSystem 探测系统能力（不需要网络目标） +func ProbeSystem() SystemProfile { + p := SystemProfile{ + NumCPU: runtime.NumCPU(), + } + p.FDLimit = getFDLimit() + return p +} + +// TuneConfig 根据探测结果调整 Config 中的参数 +// 只调整用户未显式指定的参数 +// 每个参数的推导都有明确的公式和探测依据 +func (ep *EnvironmentProfile) TuneConfig(config *common.Config, session *common.ScanSession) { + net := &ep.Net + sys := &ep.System + + // ---------- ThreadNum ---------- + // 已在 AdaptivePool 层处理（ProbeNetwork + AIMD），这里不重复 + + // ---------- Timeout ---------- + // 公式: median_rtt + 4 * stddev，下限 1s，上限 10s + // 依据: 与 AdaptiveTimeout 相同的统计原理（覆盖 99.9% 的正常连接） + if !isExplicit(config, "time") && net.Samples > 0 { + computed := net.RTTMedian + 4*net.RTTStddev + // 下限：连接建立至少需要 2 个 RTT（SYN + SYN-ACK）+ 处理时间 + minTO := net.RTTMedian*3 + 200*time.Millisecond + if computed < minTO { + computed = minTO + } + computed = clampDuration(computed, time.Second, 10*time.Second) + + old := config.Timeout + config.Timeout = computed + session.LogDebug(fmt.Sprintf("Timeout: %v -> %v (RTT median=%v stddev=%v)", + old, computed, net.RTTMedian, net.RTTStddev)) + } + + // ---------- ModuleThreadNum ---------- + // 公式: ThreadNum / 30，下限 5，上限 50 + // 依据: 插件级并发（爆破等）不应超过端口扫描并发的 ~3% + // 单个服务的连接能力远低于 TCP SYN 扫描 + // 公网服务通常有限流（MaxStartups 等），并发过高适得其反 + if !isExplicit(config, "mt") { + target, _ := net.RecommendConcurrency(config.ThreadNum, config.ThreadNumExplicit) + computed := target / 30 + computed = clampInt(computed, 5, 50) + + // 高丢包环境进一步压低，避免大量连接被丢弃浪费 + if net.LossRate > 0.1 { + computed = computed * 2 / 3 + if computed < 5 { + computed = 5 + } + } + + old := config.ModuleThreadNum + config.ModuleThreadNum = computed + session.LogDebug(fmt.Sprintf("ModuleThreadNum: %d -> %d (target_concurrency=%d)", old, computed, target)) + } + + // ---------- MaxRetries ---------- + // 公式: ceil(log(0.01) / log(loss_rate)) + // 含义: 重试 N 次后仍然全部丢包的概率 < 1% + // 例: 丢包率 5% → N=2, 丢包率 20% → N=3, 丢包率 50% → N=7 + // 下限 1（零丢包也至少试一次），上限 6（避免对不可达目标死磕） + if !isExplicit(config, "retry") && net.Samples > 0 { + computed := computeRetries(net.LossRate) + old := config.MaxRetries + config.MaxRetries = computed + session.LogDebug(fmt.Sprintf("MaxRetries: %d -> %d (loss_rate=%.2f%%)", old, computed, net.LossRate*100)) + } + + // ---------- ICMPRate ---------- + // 公式: 基于 fd limit 和网络环境 + // 内网 fd 充裕: 0.5（高速发包） + // 公网或 fd 紧张: 0.1（默认保守） + // 依据: ICMP 发包速率受两个约束：网络带宽和本机 fd/socket 资源 + if !isExplicit(config, "icmp-rate") && net.Samples > 0 { + computed := computeICMPRate(net, sys) + old := config.Network.ICMPRate + config.Network.ICMPRate = computed + session.LogDebug(fmt.Sprintf("ICMPRate: %.2f -> %.2f (env=%s fd=%d)", old, computed, net.Env, sys.FDLimit)) + } + + // ---------- PocNum ---------- + // 公式: 与 ModuleThreadNum 一致 + // 依据: POC 检测和凭据爆破的并发约束相同——都是对目标服务发起连接 + if !isExplicit(config, "num") { + old := config.POC.Num + config.POC.Num = config.ModuleThreadNum + session.LogDebug(fmt.Sprintf("PocNum: %d -> %d (follows ModuleThreadNum)", old, config.POC.Num)) + } + + // ---------- DisablePing ---------- + // 由 probeWithICMP 自动处理（尝试 → 失败 → 降级），无需在此干预 + + // 总结日志 + if net.Samples > 0 { + session.LogInfo(i18n.Tr("env_tune_summary", + config.Timeout.Milliseconds(), + config.ModuleThreadNum, + config.MaxRetries, + fmt.Sprintf("%.2f", config.Network.ICMPRate), + config.POC.Num)) + } + + // fd limit 约束：总并发不应超过 fd limit 的 60%（留余量给系统） + if sys.FDLimit > 0 { + maxConcurrency := sys.FDLimit * 6 / 10 + if config.ThreadNum > maxConcurrency { + session.LogInfo(i18n.Tr("env_fd_limit", config.ThreadNum, maxConcurrency, sys.FDLimit)) + config.ThreadNum = maxConcurrency + } + } +} + +// computeRetries 基于丢包率计算重试次数 +// 目标：重试 N 次后仍全部失败的概率 < 1% +func computeRetries(lossRate float64) int { + if lossRate <= 0.001 { + return 1 // 几乎无丢包 + } + if lossRate >= 0.95 { + return 6 // 上限 + } + // P(N次全失败) = lossRate^N < 0.01 + // N > log(0.01) / log(lossRate) + n := math.Ceil(math.Log(0.01) / math.Log(lossRate)) + return clampInt(int(n), 1, 6) +} + +// computeICMPRate 基于环境计算 ICMP 发包速率 +func computeICMPRate(net *NetworkProfile, sys *SystemProfile) float64 { + // 基准：根据 RTT 估算网络可承受的速率 + // RTT 越低，网络越快，可以发更快 + var base float64 + switch net.Env { + case EnvLAN: + base = 0.5 + case EnvWAN: + base = 0.3 + case EnvInternet: + base = 0.1 + default: + base = 0.05 + } + + // fd 约束：fd limit 低时压低速率 + if sys.FDLimit > 0 && sys.FDLimit < 1024 { + base = base * float64(sys.FDLimit) / 1024.0 + if base < 0.02 { + base = 0.02 + } + } + + return base +} + +// isExplicit 检查参数是否被用户显式指定 +// 目前只有 ThreadNum 有 explicit 标记，其他参数通过检查是否为默认值来判断 +func isExplicit(config *common.Config, flagName string) bool { + switch flagName { + case "t": + return config.ThreadNumExplicit + case "time": + return config.Timeout != 3*time.Second // 默认值 + case "mt": + return config.ModuleThreadNum != 20 // 默认值 + case "retry": + return config.MaxRetries != 3 // 默认值 + case "icmp-rate": + return config.Network.ICMPRate != 0.1 // 默认值 + case "num": + return config.POC.Num != 20 // 默认值 + } + return false +} + +func clampInt(v, min, max int) int { + if v < min { + return min + } + if v > max { + return max + } + return v +} + +func clampDuration(v, min, max time.Duration) time.Duration { + if v < min { + return min + } + if v > max { + return max + } + return v +} diff --git a/core/env_profiler_test.go b/core/env_profiler_test.go new file mode 100644 index 0000000..eebde01 --- /dev/null +++ b/core/env_profiler_test.go @@ -0,0 +1,334 @@ +package core + +import ( + "math" + "testing" + "time" + + "github.com/shadow1ng/fscan/common" +) + +// ============================================================================= +// 单元测试：computeRetries — 丢包率到重试次数的推导 +// ============================================================================= + +func TestComputeRetries(t *testing.T) { + tests := []struct { + lossRate float64 + wantMin int + wantMax int + desc string + }{ + {0.0, 1, 1, "零丢包: 只需 1 次"}, + {0.001, 1, 1, "极低丢包: 1 次"}, + {0.05, 2, 2, "5% 丢包: 0.05^2=0.0025 < 0.01"}, + {0.10, 2, 3, "10% 丢包: ceil(log(0.01)/log(0.1))=2, 但边界取 ceil 可能是 3"}, + {0.20, 3, 3, "20% 丢包: 0.2^3=0.008 < 0.01"}, + {0.30, 3, 4, "30% 丢包"}, + {0.50, 6, 6, "50% 丢包: ceil(log(0.01)/log(0.5))=7 但上限 6"}, + {0.80, 6, 6, "80% 丢包: 需要很多次但上限 6"}, + {0.95, 6, 6, "95% 丢包: 触顶"}, + {1.0, 6, 6, "100% 丢包: 触顶"}, + } + + for _, tt := range tests { + t.Run(tt.desc, func(t *testing.T) { + got := computeRetries(tt.lossRate) + if got < tt.wantMin || got > tt.wantMax { + t.Errorf("computeRetries(%.2f) = %d, want [%d, %d]", + tt.lossRate, got, tt.wantMin, tt.wantMax) + } + + // 验证数学正确性：lossRate^got < 0.01 + // 跳过：零丢包、极高丢包（触顶上限 6 时数学不满足，属于设计取舍） + if tt.lossRate > 0.001 && tt.lossRate < 0.45 { + prob := math.Pow(tt.lossRate, float64(got)) + if prob >= 0.01 { + t.Errorf("lossRate=%.2f retries=%d: P(全失败)=%.4f >= 0.01, 重试不够", + tt.lossRate, got, prob) + } + } + }) + } +} + +// ============================================================================= +// 单元测试：computeICMPRate +// ============================================================================= + +func TestComputeICMPRate(t *testing.T) { + tests := []struct { + env NetworkEnv + fdLimit int + wantMin float64 + wantMax float64 + desc string + }{ + {EnvLAN, 65536, 0.4, 0.6, "内网高 fd: 高速"}, + {EnvWAN, 65536, 0.2, 0.4, "局域网高 fd: 中速"}, + {EnvInternet, 65536, 0.05, 0.15, "公网: 保守"}, + {EnvSlow, 65536, 0.03, 0.08, "慢速: 极保守"}, + {EnvLAN, 256, 0.01, 0.2, "内网低 fd: 受限"}, + {EnvLAN, 0, 0.4, 0.6, "fd 未知: 按环境"}, + } + + for _, tt := range tests { + t.Run(tt.desc, func(t *testing.T) { + net := &NetworkProfile{Env: tt.env} + sys := &SystemProfile{FDLimit: tt.fdLimit} + got := computeICMPRate(net, sys) + if got < tt.wantMin || got > tt.wantMax { + t.Errorf("computeICMPRate(env=%v, fd=%d) = %.3f, want [%.3f, %.3f]", + tt.env, tt.fdLimit, got, tt.wantMin, tt.wantMax) + } + }) + } +} + +// ============================================================================= +// 集成测试：TuneConfig — 完整参数调整流程 +// ============================================================================= + +func TestTuneConfig_LAN(t *testing.T) { + config := makeDefaultConfig() + session := makeTestSession(config) + + ep := &EnvironmentProfile{ + Net: NetworkProfile{ + Env: EnvLAN, + RTTMin: 500 * time.Microsecond, + RTTMedian: 1 * time.Millisecond, + RTTP95: 3 * time.Millisecond, + RTTStddev: 500 * time.Microsecond, + LossRate: 0.0, + Samples: 30, + }, + System: SystemProfile{FDLimit: 65536, NumCPU: 8}, + } + + ep.TuneConfig(config, session) + + // Timeout: median(1ms) + 4*stddev(0.5ms) = 3ms → clamp to 1s 下限 + if config.Timeout < time.Second || config.Timeout > 2*time.Second { + t.Errorf("LAN Timeout = %v, 内网应该在 1-2s", config.Timeout) + } + + // MaxRetries: 零丢包 → 1 + if config.MaxRetries != 1 { + t.Errorf("LAN MaxRetries = %d, 零丢包应该是 1", config.MaxRetries) + } + + // ICMPRate: 内网应该比默认 0.1 高 + if config.Network.ICMPRate <= 0.1 { + t.Errorf("LAN ICMPRate = %.2f, 应该 > 0.1", config.Network.ICMPRate) + } + + // ModuleThreadNum: 基于 ThreadNum/30 + if config.ModuleThreadNum < 5 { + t.Errorf("LAN ModuleThreadNum = %d, 应该 >= 5", config.ModuleThreadNum) + } + + t.Logf("LAN 参数: Timeout=%v, MT=%d, Retry=%d, ICMP=%.2f, POC=%d", + config.Timeout, config.ModuleThreadNum, config.MaxRetries, config.Network.ICMPRate, config.POC.Num) +} + +func TestTuneConfig_Internet(t *testing.T) { + config := makeDefaultConfig() + session := makeTestSession(config) + + ep := &EnvironmentProfile{ + Net: NetworkProfile{ + Env: EnvInternet, + RTTMin: 50 * time.Millisecond, + RTTMedian: 100 * time.Millisecond, + RTTP95: 250 * time.Millisecond, + RTTStddev: 40 * time.Millisecond, + LossRate: 0.08, + Samples: 25, + }, + System: SystemProfile{FDLimit: 1024, NumCPU: 4}, + } + + ep.TuneConfig(config, session) + + // Timeout: median(100ms) + 4*stddev(40ms) = 260ms → 但 minTO = 3*100+200 = 500ms + if config.Timeout < 500*time.Millisecond || config.Timeout > 5*time.Second { + t.Errorf("Internet Timeout = %v, 公网应该在 500ms-5s", config.Timeout) + } + + // MaxRetries: 8% 丢包 → ceil(log(0.01)/log(0.08)) ≈ 2 + if config.MaxRetries < 2 || config.MaxRetries > 3 { + t.Errorf("Internet MaxRetries = %d, 8%%丢包应该是 2-3", config.MaxRetries) + } + + // ICMPRate: 公网应该偏低 + if config.Network.ICMPRate > 0.2 { + t.Errorf("Internet ICMPRate = %.2f, 应该 <= 0.2", config.Network.ICMPRate) + } + + t.Logf("Internet 参数: Timeout=%v, MT=%d, Retry=%d, ICMP=%.2f, POC=%d", + config.Timeout, config.ModuleThreadNum, config.MaxRetries, config.Network.ICMPRate, config.POC.Num) +} + +func TestTuneConfig_SlowLossy(t *testing.T) { + config := makeDefaultConfig() + session := makeTestSession(config) + + ep := &EnvironmentProfile{ + Net: NetworkProfile{ + Env: EnvSlow, + RTTMin: 200 * time.Millisecond, + RTTMedian: 500 * time.Millisecond, + RTTP95: 2 * time.Second, + RTTStddev: 300 * time.Millisecond, + LossRate: 0.25, + Samples: 15, + }, + System: SystemProfile{FDLimit: 512, NumCPU: 2}, + } + + ep.TuneConfig(config, session) + + // Timeout: median(500ms) + 4*stddev(300ms) = 1700ms, minTO = 500*3+200 = 1700ms + if config.Timeout < time.Second { + t.Errorf("Slow Timeout = %v, 慢速网络应该 >= 1s", config.Timeout) + } + + // MaxRetries: 25% 丢包 → ceil(log(0.01)/log(0.25)) ≈ 4 + if config.MaxRetries < 3 || config.MaxRetries > 5 { + t.Errorf("Slow MaxRetries = %d, 25%%丢包应该是 3-5", config.MaxRetries) + } + + // ICMPRate: 慢速 + 低 fd → 应该很低 + if config.Network.ICMPRate > 0.1 { + t.Errorf("Slow ICMPRate = %.2f, 应该 <= 0.1", config.Network.ICMPRate) + } + + t.Logf("Slow 参数: Timeout=%v, MT=%d, Retry=%d, ICMP=%.2f, POC=%d", + config.Timeout, config.ModuleThreadNum, config.MaxRetries, config.Network.ICMPRate, config.POC.Num) +} + +// ============================================================================= +// 集成测试：用户显式指定时不覆盖 +// ============================================================================= + +func TestTuneConfig_ExplicitOverride(t *testing.T) { + config := makeDefaultConfig() + config.Timeout = 5 * time.Second // 用户设了 -time 5 + config.ModuleThreadNum = 50 // 用户设了 -mt 50 + config.MaxRetries = 1 // 用户设了 -retry 1 + config.Network.ICMPRate = 0.8 // 用户设了 -icmp-rate 0.8 + config.POC.Num = 100 // 用户设了 -num 100 + 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.Timeout != 5*time.Second { + t.Errorf("用户 Timeout 被覆盖: %v", config.Timeout) + } + if config.ModuleThreadNum != 50 { + t.Errorf("用户 ModuleThreadNum 被覆盖: %d", config.ModuleThreadNum) + } + if config.MaxRetries != 1 { + t.Errorf("用户 MaxRetries 被覆盖: %d", config.MaxRetries) + } + if config.Network.ICMPRate != 0.8 { + t.Errorf("用户 ICMPRate 被覆盖: %.2f", config.Network.ICMPRate) + } + if config.POC.Num != 100 { + t.Errorf("用户 PocNum 被覆盖: %d", config.POC.Num) + } +} + +// ============================================================================= +// 集成测试：fd limit 约束 +// ============================================================================= + +func TestTuneConfig_FDLimitConstraint(t *testing.T) { + config := makeDefaultConfig() + config.ThreadNum = 600 + 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) + + // 600 线程 > 256 * 0.6 = 153 → 应该被约束 + maxExpected := 256 * 6 / 10 + if config.ThreadNum > maxExpected { + t.Errorf("ThreadNum = %d, 应该 <= %d (fd_limit=256)", config.ThreadNum, maxExpected) + } + + t.Logf("fd limit 约束: ThreadNum=%d (max=%d)", config.ThreadNum, maxExpected) +} + +// ============================================================================= +// 集成测试：零样本时不调整 +// ============================================================================= + +func TestTuneConfig_NoSamples(t *testing.T) { + config := makeDefaultConfig() + session := makeTestSession(config) + + origTimeout := config.Timeout + origRetry := config.MaxRetries + origICMP := config.Network.ICMPRate + + ep := &EnvironmentProfile{ + Net: NetworkProfile{Samples: 0}, + System: SystemProfile{FDLimit: 65536}, + } + + ep.TuneConfig(config, session) + + if config.Timeout != origTimeout { + t.Errorf("零样本不应改 Timeout: %v -> %v", origTimeout, config.Timeout) + } + if config.MaxRetries != origRetry { + t.Errorf("零样本不应改 MaxRetries: %d -> %d", origRetry, config.MaxRetries) + } + if config.Network.ICMPRate != origICMP { + t.Errorf("零样本不应改 ICMPRate: %.2f -> %.2f", origICMP, config.Network.ICMPRate) + } +} + +// ============================================================================= +// 辅助 +// ============================================================================= + +func makeDefaultConfig() *common.Config { + return &common.Config{ + Timeout: 3 * time.Second, + ThreadNum: 600, + ModuleThreadNum: 20, + MaxRetries: 3, + Network: common.NetworkConfig{ICMPRate: 0.1}, + POC: common.POCConfig{Num: 20}, + Output: common.OutputConfig{LogLevel: "base,info,success"}, + } +} + +func makeTestSession(config *common.Config) *common.ScanSession { + return common.NewScanSession(config, common.NewState(), &common.FlagVars{}) +} diff --git a/core/fd_limit_unix.go b/core/fd_limit_unix.go new file mode 100644 index 0000000..89f0b29 --- /dev/null +++ b/core/fd_limit_unix.go @@ -0,0 +1,13 @@ +//go:build !windows + +package core + +import "syscall" + +func getFDLimit() int { + var lim syscall.Rlimit + if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &lim); err != nil { + return 0 + } + return int(lim.Cur) +} diff --git a/core/fd_limit_windows.go b/core/fd_limit_windows.go new file mode 100644 index 0000000..033f736 --- /dev/null +++ b/core/fd_limit_windows.go @@ -0,0 +1,8 @@ +//go:build windows + +package core + +// Windows 没有 RLIMIT_NOFILE，句柄上限由系统管理 +func getFDLimit() int { + return 0 +} diff --git a/core/integration_test.go b/core/integration_test.go new file mode 100644 index 0000000..c987f46 --- /dev/null +++ b/core/integration_test.go @@ -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) + } +} diff --git a/core/network_profiler.go b/core/network_profiler.go new file mode 100644 index 0000000..6accdd3 --- /dev/null +++ b/core/network_profiler.go @@ -0,0 +1,277 @@ +package core + +import ( + "context" + "fmt" + "math" + "net" + "sort" + "sync" + "time" + + "github.com/shadow1ng/fscan/common" + "github.com/shadow1ng/fscan/common/i18n" +) + +// NetworkEnv 网络环境分类 +type NetworkEnv int + +const ( + EnvLAN NetworkEnv = iota // 内网: RTT < 5ms, 丢包 < 1% + EnvWAN // 局域网/专线: RTT 5~50ms, 丢包 < 5% + EnvInternet // 公网: RTT 50~200ms + EnvSlow // 慢速/高丢包: RTT > 200ms 或 丢包 > 10% +) + +func (e NetworkEnv) String() string { + switch e { + case EnvLAN: + return i18n.GetText("net_env_lan") + case EnvWAN: + return i18n.GetText("net_env_wan") + case EnvInternet: + return i18n.GetText("net_env_internet") + default: + return i18n.GetText("net_env_slow") + } +} + +// NetworkProfile 网络探测结果 +type NetworkProfile struct { + Env NetworkEnv + RTTMin time.Duration + RTTMedian time.Duration + RTTP95 time.Duration + RTTStddev time.Duration + LossRate float64 + Samples int +} + +// RecommendConcurrency 根据探测结果推荐并发参数 +// 返回 (target, ceiling) +// - target: 推荐的目标并发数 +// - ceiling: 允许的最大并发数 +// +// 如果用户显式指定了 -t，ceiling = 用户值，target 取 min(推荐值, 用户值) +// 如果用户未指定，target 和 ceiling 均为推荐值 +func (p *NetworkProfile) RecommendConcurrency(userThreadNum int, explicit bool) (target, ceiling int) { + // 基于网络环境的缩放因子 + var factor float64 + switch p.Env { + case EnvLAN: + factor = 1.5 + case EnvWAN: + factor = 1.0 + case EnvInternet: + factor = 0.4 + case EnvSlow: + factor = 0.15 + } + + recommended := int(float64(userThreadNum) * factor) + if recommended < 10 { + recommended = 10 + } + + // 丢包率高时进一步压缩 + if p.LossRate > 0.05 { + recommended = int(float64(recommended) * (1.0 - p.LossRate)) + if recommended < 10 { + recommended = 10 + } + } + + if explicit { + ceiling = userThreadNum + target = recommended + if target > ceiling { + target = ceiling + } + } else { + target = recommended + ceiling = recommended + } + return +} + +// probePorts 探测用的端口列表（高响应率的常见端口） +var probePorts = []int{80, 443, 22} + +// ProbeNetwork 探测目标网络环境 +// 从 hosts 中抽样，用低并发 TCP 连接测量 RTT 和丢包率 +// 整个过程控制在数秒内完成 +func ProbeNetwork(ctx context.Context, hosts []string, session *common.ScanSession) *NetworkProfile { + if len(hosts) == 0 { + return defaultProfile() + } + + // 抽样：均匀分布，最多 10 个 + samples := pickSamples(hosts, 10) + probeTimeout := session.Config.Timeout + if probeTimeout > time.Second { + probeTimeout = time.Second + } + if probeTimeout < 500*time.Millisecond { + probeTimeout = 500 * time.Millisecond + } + + var ( + mu sync.Mutex + rtts []time.Duration + failures int + total int + ) + + sem := make(chan struct{}, 10) + var wg sync.WaitGroup + + for _, host := range samples { + for _, port := range probePorts { + select { + case <-ctx.Done(): + goto done + default: + } + + total++ + wg.Add(1) + sem <- struct{}{} + + go func(h string, p int) { + defer func() { <-sem; wg.Done() }() + + addr := fmt.Sprintf("%s:%d", h, p) + start := time.Now() + conn, err := session.DialTCP(ctx, "tcp", addr, probeTimeout) + rtt := time.Since(start) + + mu.Lock() + defer mu.Unlock() + + if err != nil { + // 连接拒绝也是有效的 RTT 样本（说明对端可达） + if isConnectionRefused(err) { + rtts = append(rtts, rtt) + } + failures++ + } else { + _ = conn.Close() + rtts = append(rtts, rtt) + } + }(host, port) + } + } +done: + wg.Wait() + + return classifyNetwork(rtts, failures, total) +} + +func classifyNetwork(rtts []time.Duration, failures, total int) *NetworkProfile { + if len(rtts) == 0 { + return defaultProfile() + } + + sort.Slice(rtts, func(i, j int) bool { return rtts[i] < rtts[j] }) + + n := len(rtts) + median := rtts[n/2] + p95idx := int(float64(n) * 0.95) + if p95idx >= n { + p95idx = n - 1 + } + p95 := rtts[p95idx] + + // 标准差 + var sum float64 + for _, r := range rtts { + sum += float64(r) + } + mean := sum / float64(n) + var variance float64 + for _, r := range rtts { + d := float64(r) - mean + variance += d * d + } + stddev := time.Duration(math.Sqrt(variance / float64(n))) + + // 丢包率：只计算超时的（非 refused），但简化为 1 - 有效响应数/总数 + lossRate := 1.0 - float64(n)/float64(total) + if lossRate < 0 { + lossRate = 0 + } + + // 分类 + env := classifyEnv(median, lossRate) + + return &NetworkProfile{ + Env: env, + RTTMin: rtts[0], + RTTMedian: median, + RTTP95: p95, + RTTStddev: stddev, + LossRate: lossRate, + Samples: n, + } +} + +func classifyEnv(median time.Duration, lossRate float64) NetworkEnv { + switch { + case lossRate > 0.10: + return EnvSlow + case median < 5*time.Millisecond && lossRate < 0.01: + return EnvLAN + case median < 50*time.Millisecond && lossRate < 0.05: + return EnvWAN + case median < 200*time.Millisecond: + return EnvInternet + default: + return EnvSlow + } +} + +func defaultProfile() *NetworkProfile { + return &NetworkProfile{ + Env: EnvWAN, + RTTMedian: 10 * time.Millisecond, + LossRate: 0, + Samples: 0, + } +} + +// pickSamples 均匀抽样 +func pickSamples(hosts []string, maxSamples int) []string { + if maxSamples <= 0 { + return nil + } + n := len(hosts) + if n <= maxSamples { + return hosts + } + step := n / maxSamples + samples := make([]string, 0, maxSamples) + for i := 0; i < n && len(samples) < maxSamples; i += step { + samples = append(samples, hosts[i]) + } + return samples +} + +func isConnectionRefused(err error) bool { + if err == nil { + return false + } + // connection refused 通常包含 "refused" 关键词 + // 在不同 OS 上表现一致 + return containsFold(err.Error(), "refused") +} + +// isTimeoutError 判断是否为超时错误 +func isTimeoutError(err error) bool { + if err == nil { + return false + } + if ne, ok := err.(net.Error); ok { + return ne.Timeout() + } + return containsFold(err.Error(), "timeout") || containsFold(err.Error(), "deadline") +} diff --git a/core/network_profiler_test.go b/core/network_profiler_test.go new file mode 100644 index 0000000..252c385 --- /dev/null +++ b/core/network_profiler_test.go @@ -0,0 +1,169 @@ +package core + +import ( + "testing" + "time" +) + +// ============================================================================= +// 单元测试：classifyEnv — 网络环境分类 +// ============================================================================= + +func TestClassifyEnv(t *testing.T) { + tests := []struct { + median time.Duration + lossRate float64 + wantEnv NetworkEnv + desc string + }{ + {1 * time.Millisecond, 0.0, EnvLAN, "1ms 零丢包 → 内网"}, + {3 * time.Millisecond, 0.005, EnvLAN, "3ms 0.5%丢包 → 内网"}, + {5 * time.Millisecond, 0.0, EnvWAN, "5ms 零丢包 → 局域网边界"}, + {20 * time.Millisecond, 0.02, EnvWAN, "20ms 2%丢包 → 局域网"}, + {50 * time.Millisecond, 0.03, EnvInternet, "50ms 3%丢包 → 公网边界"}, + {100 * time.Millisecond, 0.05, EnvInternet, "100ms 5%丢包 → 公网"}, + {300 * time.Millisecond, 0.05, EnvSlow, "300ms → 慢速"}, + {50 * time.Millisecond, 0.15, EnvSlow, "50ms 15%丢包 → 高丢包归类慢速"}, + {1 * time.Millisecond, 0.20, EnvSlow, "低延迟但高丢包 → 慢速"}, + } + + for _, tt := range tests { + t.Run(tt.desc, func(t *testing.T) { + got := classifyEnv(tt.median, tt.lossRate) + if got != tt.wantEnv { + t.Errorf("classifyEnv(median=%v, loss=%.2f) = %v, want %v", + tt.median, tt.lossRate, got, tt.wantEnv) + } + }) + } +} + +// ============================================================================= +// 单元测试：classifyNetwork — 从 RTT 样本推导 profile +// ============================================================================= + +func TestClassifyNetwork(t *testing.T) { + t.Run("内网 RTT 分布", func(t *testing.T) { + rtts := makeDurations([]int{1, 1, 1, 2, 2, 2, 3, 3, 4, 5}) // ms + p := classifyNetwork(rtts, 0, 10) + + if p.Env != EnvLAN { + t.Errorf("env = %v, want LAN", p.Env) + } + if p.RTTMedian > 5*time.Millisecond { + t.Errorf("median = %v, want < 5ms", p.RTTMedian) + } + if p.LossRate != 0 { + t.Errorf("lossRate = %.2f, want 0", p.LossRate) + } + }) + + t.Run("公网 RTT 分布（低丢包）", func(t *testing.T) { + rtts := makeDurations([]int{60, 70, 80, 90, 100, 110, 120, 150, 200, 300}) // ms + p := classifyNetwork(rtts, 0, 10) // 无丢包 + + if p.Env != EnvInternet { + t.Errorf("env = %v, want Internet", p.Env) + } + if p.LossRate != 0 { + t.Errorf("lossRate = %.2f, want 0", p.LossRate) + } + }) + + t.Run("高丢包归类为慢速", func(t *testing.T) { + rtts := makeDurations([]int{60, 70, 80, 90, 100}) // ms, 5 responded + p := classifyNetwork(rtts, 5, 10) // 50% loss + + if p.Env != EnvSlow { + t.Errorf("env = %v, want Slow (高丢包)", p.Env) + } + }) + + t.Run("零样本降级", func(t *testing.T) { + p := classifyNetwork(nil, 5, 5) + if p.Env != EnvWAN { + t.Errorf("env = %v, want WAN (default)", p.Env) + } + if p.Samples != 0 { + t.Errorf("samples = %d, want 0", p.Samples) + } + }) +} + +// ============================================================================= +// 单元测试：RecommendConcurrency +// ============================================================================= + +func TestRecommendConcurrency(t *testing.T) { + tests := []struct { + env NetworkEnv + lossRate float64 + userT int + explicit bool + wantTMin int + wantTMax int + wantCeil int + desc string + }{ + {EnvLAN, 0.0, 600, false, 800, 1000, -1, "内网自动: ×1.5"}, + {EnvWAN, 0.0, 600, false, 550, 650, -1, "局域网自动: ×1.0"}, + {EnvInternet, 0.0, 600, false, 200, 280, -1, "公网自动: ×0.4"}, + {EnvSlow, 0.0, 600, false, 80, 100, -1, "慢速自动: ×0.15"}, + {EnvInternet, 0.0, 200, true, 70, 100, 200, "公网显式: target tt.wantTMax { + t.Errorf("target = %d, want [%d, %d]", target, tt.wantTMin, tt.wantTMax) + } + + if tt.explicit && ceiling != tt.wantCeil { + t.Errorf("ceiling = %d, want %d", ceiling, tt.wantCeil) + } + }) + } +} + +// ============================================================================= +// 单元测试：pickSamples +// ============================================================================= + +func TestPickSamples(t *testing.T) { + hosts := make([]string, 100) + for i := range hosts { + hosts[i] = "host" + } + + s := pickSamples(hosts, 10) + if len(s) != 10 { + t.Errorf("pickSamples(100, 10) = %d items, want 10", len(s)) + } + + s = pickSamples(hosts[:5], 10) + if len(s) != 5 { + t.Errorf("pickSamples(5, 10) = %d items, want 5", len(s)) + } + + s = pickSamples(nil, 10) + if len(s) != 0 { + t.Errorf("pickSamples(nil, 10) = %d items, want 0", len(s)) + } +} + +// ============================================================================= +// 辅助 +// ============================================================================= + +func makeDurations(ms []int) []time.Duration { + ds := make([]time.Duration, len(ms)) + for i, m := range ms { + ds[i] = time.Duration(m) * time.Millisecond + } + return ds +} diff --git a/core/port_scan.go b/core/port_scan.go index 0d37153..150d717 100644 --- a/core/port_scan.go +++ b/core/port_scan.go @@ -187,13 +187,12 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout totalTasks := iter.Total() session.LogDebug(i18n.Tr("port_scan_debug_total_tasks", totalTasks)) - // 使用传入的配置 + // 并发参数（已由 EnvironmentProfile.TuneConfig 调整过） threadNum := config.ThreadNum - // 大规模扫描警告和线程数自动调整 + // 大规模扫描额外约束 if totalTasks > 100000 { session.LogInfo(i18n.Tr("large_scan_notice", totalTasks, len(hosts), len(portList))) - // 如果任务数超过100万且线程数大于300，自动降低线程数 if totalTasks > 1000000 && threadNum > 300 { oldThreadNum := threadNum threadNum = 300 @@ -211,14 +210,14 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout // 初始化并发控制 to := time.Duration(timeout) * time.Second adaptiveTO := NewAdaptiveTimeout(to) + metrics := &ScanMetrics{} var count atomic.Int64 collector := newResultCollector(stream) failedCollector := &failedPortCollector{} var wg sync.WaitGroup session.LogDebug(i18n.Tr("port_scan_debug_pool_create", threadNum)) - // 创建自适应线程池（支持动态调整） - pool, err := NewAdaptivePool(threadNum, func(task interface{}) { + pool, err := NewAdaptivePool(threadNum, threadNum, func(task interface{}) { taskInfo, ok := task.(portScanTask) if !ok { return @@ -228,9 +227,9 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout wg.Done() }() - scanSinglePort(ctx, taskInfo.host, taskInfo.port, taskInfo.addr, adaptiveTO, &count, collector, failedCollector, session) + scanSinglePort(ctx, taskInfo.host, taskInfo.port, taskInfo.addr, adaptiveTO, metrics, &count, collector, failedCollector, session) common.UpdateProgressBar(1) - }, state) + }, metrics) if err != nil { session.LogError(i18n.Tr("thread_pool_create_failed", err)) if stream != nil { @@ -242,7 +241,7 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout defer pool.Release() session.LogDebug(i18n.GetText("port_scan_debug_schedule_start")) - // 滑动窗口调度：维护固定数量的"飞行中"任务 + // 滑动窗口调度 slidingWindowSchedule(iter, pool, &wg, threadNum) session.LogDebug(i18n.GetText("port_scan_debug_schedule_done")) @@ -482,17 +481,28 @@ func buildWebServiceURL(addr string, serviceInfo *ServiceInfo) string { } // scanSinglePort 扫描单个端口并进行服务识别（重构后的简洁版本） -func scanSinglePort(ctx context.Context, host string, port int, addr string, adaptiveTO *AdaptiveTimeout, count *atomic.Int64, collector *resultCollector, failedCollector *failedPortCollector, session *common.ScanSession) { +func scanSinglePort(ctx context.Context, host string, port int, addr string, adaptiveTO *AdaptiveTimeout, metrics *ScanMetrics, count *atomic.Int64, collector *resultCollector, failedCollector *failedPortCollector, session *common.ScanSession) { config := session.Config timeout := adaptiveTO.Timeout() // 步骤1：建立连接 start := time.Now() conn, err := connectWithRetry(ctx, session, addr, timeout, 2) if err != nil { + rtt := time.Since(start) + switch { + case isResourceExhaustedError(err): + metrics.RecordExhausted() + case isTimeoutError(err): + metrics.RecordTimeout() + default: + metrics.RecordRefused(rtt) + } handleConnectionFailure(err, host, port, addr, failedCollector) return } - adaptiveTO.Record(time.Since(start)) + rtt := time.Since(start) + metrics.RecordConnect(rtt) + adaptiveTO.Record(rtt) // 步骤1.5：代理连接深度验证（防止透明代理/全回显代理的假连接问题） valid, verifyMethod := verifyProxyConnectionDeep(conn, addr, session) diff --git a/core/real_network_test.go b/core/real_network_test.go new file mode 100644 index 0000000..74b396b --- /dev/null +++ b/core/real_network_test.go @@ -0,0 +1,487 @@ +package core + +import ( + "context" + "fmt" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/shadow1ng/fscan/common" +) + +// ============================================================================= +// 辅助：启动本地 TCP 监听器 +// ============================================================================= + +// startListeners 启动 N 个本地 TCP 监听端口，返回地址列表和清理函数 +func startListeners(t *testing.T, n int) (addrs []string, hosts []string, ports []int, cleanup func()) { + t.Helper() + var listeners []net.Listener + + for i := 0; i < n; i++ { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + for _, l := range listeners { + l.Close() + } + t.Fatalf("启动监听失败: %v", err) + } + listeners = append(listeners, ln) + addr := ln.Addr().String() + addrs = append(addrs, addr) + + host, portStr, _ := net.SplitHostPort(addr) + hosts = append(hosts, host) + var port int + fmt.Sscanf(portStr, "%d", &port) + ports = append(ports, port) + + // 后台 accept（不处理连接，只让 connect 成功） + go func(l net.Listener) { + for { + conn, err := l.Accept() + if err != nil { + return + } + conn.Close() + } + }(ln) + } + + return addrs, hosts, ports, func() { + for _, l := range listeners { + l.Close() + } + } +} + +// makeRealSession 创建用于真实网络测试的 session +func makeRealSession(t *testing.T) (*common.Config, *common.ScanSession) { + t.Helper() + config := &common.Config{ + Timeout: 3 * time.Second, + ThreadNum: 100, + ModuleThreadNum: 10, + MaxRetries: 3, + Network: common.NetworkConfig{ICMPRate: 0.1}, + POC: common.POCConfig{Num: 20}, + Output: common.OutputConfig{LogLevel: "base,info,success"}, + } + session := common.NewScanSession(config, common.NewState(), &common.FlagVars{}) + return config, session +} + +// ============================================================================= +// 真实测试 1：ProbeNetwork 对 localhost 探测 +// ============================================================================= + +func TestReal_ProbeNetwork_Localhost(t *testing.T) { + _, hosts, _, cleanup := startListeners(t, 3) + defer cleanup() + + _, session := makeRealSession(t) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + profile := ProbeNetwork(ctx, hosts, session) + + if profile.Samples == 0 { + t.Fatal("localhost 探测应该有样本") + } + + // localhost 应该是内网环境 + if profile.Env != EnvLAN { + t.Errorf("localhost env = %v, want LAN", profile.Env) + } + + // RTT 应该 < 10ms + if profile.RTTMedian > 10*time.Millisecond { + t.Errorf("localhost RTT median = %v, 应该 < 10ms", profile.RTTMedian) + } + + // 丢包率应该为 0 或极低 + if profile.LossRate > 0.1 { + t.Errorf("localhost loss = %.2f, 应该接近 0", profile.LossRate) + } + + t.Logf("localhost 探测: env=%v RTT_median=%v RTT_p95=%v loss=%.2f%% samples=%d", + profile.Env, profile.RTTMedian, profile.RTTP95, profile.LossRate*100, profile.Samples) +} + +// ============================================================================= +// 真实测试 2：ProbeNetwork 对不可达目标 +// ============================================================================= + +func TestReal_ProbeNetwork_Unreachable(t *testing.T) { + _, session := makeRealSession(t) + // 使用 RFC 5737 保留地址段，保证不可达 + hosts := []string{"192.0.2.1", "192.0.2.2", "192.0.2.3"} + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + profile := ProbeNetwork(ctx, hosts, session) + + // 不可达目标应该返回默认 profile 或高丢包 + t.Logf("不可达探测: env=%v samples=%d loss=%.2f%%", + profile.Env, profile.Samples, profile.LossRate*100) +} + +// ============================================================================= +// 真实测试 3：ProbeNetwork 混合可达与不可达 +// ============================================================================= + +func TestReal_ProbeNetwork_Mixed(t *testing.T) { + _, hosts, _, cleanup := startListeners(t, 2) + defer cleanup() + + // 混合真实主机和不可达地址 + mixed := append(hosts, "192.0.2.1", "192.0.2.2") + + _, session := makeRealSession(t) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + profile := ProbeNetwork(ctx, mixed, session) + + if profile.Samples == 0 { + t.Error("混合探测应该有一些成功样本") + } + + t.Logf("混合探测: env=%v RTT=%v samples=%d loss=%.2f%%", + profile.Env, profile.RTTMedian, profile.Samples, profile.LossRate*100) +} + +// ============================================================================= +// 真实测试 4：ProbeSystem +// ============================================================================= + +func TestReal_ProbeSystem(t *testing.T) { + sys := ProbeSystem() + + if sys.NumCPU <= 0 { + t.Errorf("NumCPU = %d, 应该 > 0", sys.NumCPU) + } + + t.Logf("系统探测: NumCPU=%d FDLimit=%d", sys.NumCPU, sys.FDLimit) + + // Linux/macOS 上 FDLimit 应该 > 0 + // Windows 上可能为 0（设计如此） + if sys.FDLimit < 0 { + t.Errorf("FDLimit = %d, 不应为负", sys.FDLimit) + } +} + +// ============================================================================= +// 真实测试 5：完整链路 —— 探测 → 调参 → 池创建 → 真实任务执行 +// ============================================================================= + +func TestReal_E2E_ProbeAndScan(t *testing.T) { + addrs, hosts, _, cleanup := startListeners(t, 5) + defer cleanup() + + config, session := makeRealSession(t) + + // 第一步：探测 + ctx := context.Background() + profile := ProbeNetwork(ctx, hosts, session) + sys := ProbeSystem() + ep := &EnvironmentProfile{Net: *profile, System: sys} + + // 第二步：调参 + ep.TuneConfig(config, session) + + // 第三步：创建池 + target, ceiling := profile.RecommendConcurrency(config.ThreadNum, false) + metrics := &ScanMetrics{} + + var successCount atomic.Int64 + + pool, err := NewAdaptivePool(target, ceiling, func(i interface{}) { + addr := i.(string) + conn, err := net.DialTimeout("tcp", addr, config.Timeout) + if err != nil { + metrics.RecordTimeout() + return + } + defer conn.Close() + successCount.Add(1) + metrics.RecordConnect(time.Millisecond) + }, metrics) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + // 跳过慢启动测试主要流程 + pool.inSlowStart = false + pool.tune(target) + + // 第四步：提交任务 + var wg sync.WaitGroup + for _, addr := range addrs { + wg.Add(1) + a := addr + go func() { + defer wg.Done() + _ = pool.Invoke(a) + }() + } + wg.Wait() + pool.Wait() + + // 第五步：验证 + if successCount.Load() != int64(len(addrs)) { + t.Errorf("成功连接 %d/%d", successCount.Load(), len(addrs)) + } + + snap := metrics.Snapshot() + if snap.Connects != int64(len(addrs)) { + t.Errorf("metrics.Connects = %d, want %d", snap.Connects, len(addrs)) + } + + t.Logf("E2E: profile=%v timeout=%v mt=%d retry=%d target=%d connects=%d", + profile.Env, config.Timeout, config.ModuleThreadNum, config.MaxRetries, + target, snap.Connects) +} + +// ============================================================================= +// 真实测试 6：大量连接的自适应行为 +// ============================================================================= + +func TestReal_AdaptivePool_ManyConnections(t *testing.T) { + _, hosts, ports, cleanup := startListeners(t, 3) + defer cleanup() + + metrics := &ScanMetrics{} + var successCount, failCount atomic.Int64 + + pool, err := NewAdaptivePool(50, 50, func(i interface{}) { + addr := i.(string) + start := time.Now() + conn, err := net.DialTimeout("tcp", addr, time.Second) + rtt := time.Since(start) + if err != nil { + failCount.Add(1) + metrics.RecordTimeout() + return + } + defer conn.Close() + successCount.Add(1) + metrics.RecordConnect(rtt) + }, metrics) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + pool.inSlowStart = false + pool.tune(50) + + // 提交 300 个连接任务（对 3 个端口各 100 次） + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + for j, host := range hosts { + addr := fmt.Sprintf("%s:%d", host, ports[j]) + wg.Add(1) + go func(a string) { + defer wg.Done() + _ = pool.Invoke(a) + }(addr) + } + } + wg.Wait() + pool.Wait() + + total := successCount.Load() + failCount.Load() + if total != 300 { + t.Errorf("总任务 %d, want 300", total) + } + + snap := metrics.Snapshot() + t.Logf("大量连接: success=%d fail=%d connects=%d timeouts=%d cap=%d rtt_ratio=%.2f", + successCount.Load(), failCount.Load(), snap.Connects, snap.Timeouts, pool.Cap(), metrics.RTTRatio()) + + // localhost 连接应该几乎全部成功 + if successCount.Load() < 280 { + t.Errorf("localhost 成功率过低: %d/300", successCount.Load()) + } +} + +// ============================================================================= +// 真实测试 7：连接关闭端口 + 开放端口混合 +// ============================================================================= + +func TestReal_MixedOpenClosed(t *testing.T) { + _, hosts, ports, cleanup := startListeners(t, 2) + defer cleanup() + + metrics := &ScanMetrics{} + + pool, err := NewAdaptivePool(20, 20, func(i interface{}) { + addr := i.(string) + start := time.Now() + conn, err := net.DialTimeout("tcp", addr, time.Second) + rtt := time.Since(start) + if err != nil { + if isConnectionRefused(err) { + metrics.RecordRefused(rtt) + } else { + metrics.RecordTimeout() + } + return + } + defer conn.Close() + metrics.RecordConnect(rtt) + }, metrics) + if err != nil { + t.Fatalf("创建池失败: %v", err) + } + defer pool.Release() + + pool.inSlowStart = false + pool.tune(20) + + var wg sync.WaitGroup + + // 连接开放端口 + for i := 0; i < 20; i++ { + addr := fmt.Sprintf("%s:%d", hosts[0], ports[0]) + wg.Add(1) + go func(a string) { + defer wg.Done() + _ = pool.Invoke(a) + }(addr) + } + + // 连接关闭端口（用一个不存在的端口） + for i := 0; i < 20; i++ { + addr := fmt.Sprintf("127.0.0.1:%d", 1) // port 1 通常关闭 + wg.Add(1) + go func(a string) { + defer wg.Done() + _ = pool.Invoke(a) + }(addr) + } + + wg.Wait() + pool.Wait() + + snap := metrics.Snapshot() + t.Logf("混合端口: connects=%d refused=%d timeouts=%d total=%d", + snap.Connects, snap.Refused, snap.Timeouts, snap.Total()) + + // 开放端口应该全部连接成功 + if snap.Connects < 18 { + t.Errorf("开放端口连接数 = %d, 应该接近 20", snap.Connects) + } + + // RTT ratio 应该合理（不会因为 refused 而异常） + ratio := metrics.RTTRatio() + if ratio > 3.0 || ratio < 0.3 { + t.Errorf("混合流量 RTT ratio = %.2f, 不合理", ratio) + } +} + +// ============================================================================= +// 真实测试 8：Context 取消时的探测行为 +// ============================================================================= + +func TestReal_ProbeNetwork_ContextCancel(t *testing.T) { + _, hosts, _, cleanup := startListeners(t, 3) + defer cleanup() + + _, session := makeRealSession(t) + + // 立即取消的 context + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + profile := ProbeNetwork(ctx, hosts, session) + + // 应该优雅返回默认 profile 或部分结果 + t.Logf("取消探测: env=%v samples=%d", profile.Env, profile.Samples) +} + +// ============================================================================= +// 真实测试 9：AdaptiveTimeout 真实 RTT 收敛 +// ============================================================================= + +func TestReal_AdaptiveTimeout_Convergence(t *testing.T) { + addrs, _, _, cleanup := startListeners(t, 1) + defer cleanup() + + at := NewAdaptiveTimeout(3 * time.Second) + + // 初始应该返回最大超时 + if at.Timeout() != 3*time.Second { + t.Errorf("冷启动 Timeout = %v, want 3s", at.Timeout()) + } + + // 做 20 次真实连接采样 + for i := 0; i < 20; i++ { + start := time.Now() + conn, err := net.DialTimeout("tcp", addrs[0], time.Second) + rtt := time.Since(start) + if err != nil { + t.Fatalf("连接失败: %v", err) + } + conn.Close() + at.Record(rtt) + } + + // 采样够后 Timeout 应远小于 3s（localhost RTT 通常 < 1ms） + converged := at.Timeout() + if converged >= 3*time.Second { + t.Errorf("采样后 Timeout = %v, 应该 < 3s", converged) + } + if converged < 100*time.Millisecond { + t.Logf("Timeout 收敛到 %v（localhost，正常）", converged) + } + + t.Logf("AdaptiveTimeout 收敛: 3s -> %v (%d 个样本)", converged, 20) +} + +// ============================================================================= +// 真实测试 10：完整 TuneConfig 对真实探测数据 +// ============================================================================= + +func TestReal_TuneConfig_WithRealProbe(t *testing.T) { + _, hosts, _, cleanup := startListeners(t, 5) + defer cleanup() + + config, session := makeRealSession(t) + + ctx := context.Background() + profile := ProbeNetwork(ctx, hosts, session) + sys := ProbeSystem() + + origTimeout := config.Timeout + origMT := config.ModuleThreadNum + origRetry := config.MaxRetries + origICMP := config.Network.ICMPRate + + ep := &EnvironmentProfile{Net: *profile, System: sys} + ep.TuneConfig(config, session) + + t.Logf("真实调参:") + t.Logf(" Timeout: %v -> %v", origTimeout, config.Timeout) + t.Logf(" MT: %d -> %d", origMT, config.ModuleThreadNum) + t.Logf(" Retry: %d -> %d", origRetry, config.MaxRetries) + t.Logf(" ICMPRate: %.2f -> %.2f", origICMP, config.Network.ICMPRate) + t.Logf(" PocNum: 20 -> %d", config.POC.Num) + t.Logf(" ThreadNum: %d (fd_limit=%d)", config.ThreadNum, sys.FDLimit) + + // localhost 环境下的基本验证 + if config.Timeout > 3*time.Second { + t.Errorf("localhost Timeout = %v, 不应高于默认 3s", config.Timeout) + } + if config.MaxRetries > 3 { + t.Errorf("localhost Retry = %d, 不应高于默认 3", config.MaxRetries) + } +} diff --git a/core/scan_metrics.go b/core/scan_metrics.go new file mode 100644 index 0000000..557e1b3 --- /dev/null +++ b/core/scan_metrics.go @@ -0,0 +1,115 @@ +package core + +import ( + "sync/atomic" + "time" +) + +// ScanMetrics 扫描过程中的实时度量指标 +// 所有方法均无锁，使用 atomic 操作，可在高并发下安全调用 +type ScanMetrics struct { + connects atomic.Int64 // TCP 连接成功（端口开放） + refused atomic.Int64 // 连接被拒绝（端口关闭，快速 RTT） + timeouts atomic.Int64 // 连接超时（端口过滤/不可达） + exhausted atomic.Int64 // 资源耗尽（fd/端口/内存不足） + + // RTT 追踪：双 EMA（指数移动平均） + // fast EMA (α=0.1) 跟踪近期趋势 + // slow EMA (α=0.02) 作为基线参考 + rttFastNs atomic.Int64 // 纳秒 + rttSlowNs atomic.Int64 // 纳秒 + rttSamples atomic.Int64 +} + +func (m *ScanMetrics) RecordConnect(rtt time.Duration) { + m.connects.Add(1) + m.recordRTT(rtt) +} + +func (m *ScanMetrics) RecordRefused(rtt time.Duration) { + m.refused.Add(1) + m.recordRTT(rtt) +} + +func (m *ScanMetrics) RecordTimeout() { m.timeouts.Add(1) } +func (m *ScanMetrics) RecordExhausted() { m.exhausted.Add(1) } + +// recordRTT 更新 RTT 双 EMA（lock-free CAS） +func (m *ScanMetrics) recordRTT(rtt time.Duration) { + ns := int64(rtt) + if ns <= 0 { + return + } + m.rttSamples.Add(1) + + // Fast EMA: α = 0.1 → new = old + (sample - old) / 10 + updateEMA(&m.rttFastNs, ns, 10) + // Slow EMA: α = 0.02 → new = old + (sample - old) / 50 + updateEMA(&m.rttSlowNs, ns, 50) +} + +func updateEMA(target *atomic.Int64, sample int64, divisor int64) { + for { + old := target.Load() + if old == 0 { + if target.CompareAndSwap(0, sample) { + return + } + continue + } + next := old + (sample-old)/divisor + if target.CompareAndSwap(old, next) { + return + } + } +} + +// Total 总操作数 +func (m *ScanMetrics) Total() int64 { + return m.connects.Load() + m.refused.Load() + m.timeouts.Load() + m.exhausted.Load() +} + +// MetricsSnapshot 度量快照，用于计算窗口内增量 +type MetricsSnapshot struct { + Connects int64 + Refused int64 + Timeouts int64 + Exhausted int64 + RTTFastNs int64 + RTTSlowNs int64 +} + +func (s MetricsSnapshot) Total() int64 { + return s.Connects + s.Refused + s.Timeouts + s.Exhausted +} + +func (m *ScanMetrics) Snapshot() MetricsSnapshot { + return MetricsSnapshot{ + Connects: m.connects.Load(), + Refused: m.refused.Load(), + Timeouts: m.timeouts.Load(), + Exhausted: m.exhausted.Load(), + RTTFastNs: m.rttFastNs.Load(), + RTTSlowNs: m.rttSlowNs.Load(), + } +} + +// RTTRatio 返回 fast/slow EMA 的比值 +// > 1.0 表示延迟在上升（拥塞信号），< 1.0 表示延迟在下降 +// 样本不足时返回 1.0 +func (m *ScanMetrics) RTTRatio() float64 { + if m.rttSamples.Load() < 20 { + return 1.0 + } + fast := m.rttFastNs.Load() + slow := m.rttSlowNs.Load() + if slow <= 0 { + return 1.0 + } + return float64(fast) / float64(slow) +} + +// RTTFast 返回快速 EMA 值 +func (m *ScanMetrics) RTTFast() time.Duration { + return time.Duration(m.rttFastNs.Load()) +} diff --git a/core/scan_metrics_test.go b/core/scan_metrics_test.go new file mode 100644 index 0000000..3674fd1 --- /dev/null +++ b/core/scan_metrics_test.go @@ -0,0 +1,130 @@ +package core + +import ( + "sync" + "testing" + "time" +) + +// ============================================================================= +// 单元测试：ScanMetrics 基本操作 +// ============================================================================= + +func TestScanMetrics_Counters(t *testing.T) { + m := &ScanMetrics{} + + m.RecordConnect(time.Millisecond) + m.RecordConnect(2 * time.Millisecond) + m.RecordRefused(500 * time.Microsecond) + m.RecordTimeout() + m.RecordExhausted() + + if m.Total() != 5 { + t.Errorf("Total() = %d, want 5", m.Total()) + } + + snap := m.Snapshot() + if snap.Connects != 2 { + t.Errorf("Connects = %d, want 2", snap.Connects) + } + if snap.Refused != 1 { + t.Errorf("Refused = %d, want 1", snap.Refused) + } + if snap.Timeouts != 1 { + t.Errorf("Timeouts = %d, want 1", snap.Timeouts) + } + if snap.Exhausted != 1 { + t.Errorf("Exhausted = %d, want 1", snap.Exhausted) + } +} + +// ============================================================================= +// 单元测试：RTT EMA 收敛 +// ============================================================================= + +func TestScanMetrics_RTT_EMA(t *testing.T) { + m := &ScanMetrics{} + + // 喂入稳定的 10ms RTT + for i := 0; i < 100; i++ { + m.RecordConnect(10 * time.Millisecond) + } + + fast := m.RTTFast() + if fast < 9*time.Millisecond || fast > 11*time.Millisecond { + t.Errorf("稳定 10ms 后 RTTFast = %v, 应该接近 10ms", fast) + } + + ratio := m.RTTRatio() + if ratio < 0.9 || ratio > 1.1 { + t.Errorf("稳定状态 RTTRatio = %.2f, 应该接近 1.0", ratio) + } +} + +func TestScanMetrics_RTT_Trend(t *testing.T) { + m := &ScanMetrics{} + + // 先喂入 100 个 5ms 建立基线 + for i := 0; i < 100; i++ { + m.RecordConnect(5 * time.Millisecond) + } + + // 再喂入 50 个 50ms（RTT 突增 10 倍） + for i := 0; i < 50; i++ { + m.RecordConnect(50 * time.Millisecond) + } + + ratio := m.RTTRatio() + // fast EMA 应该比 slow EMA 高（fast 跟踪快，slow 还没追上来） + if ratio <= 1.0 { + t.Errorf("RTT 突增后 RTTRatio = %.2f, 应该 > 1.0", ratio) + } + + t.Logf("RTT 突增后: ratio=%.2f, fast=%v", ratio, m.RTTFast()) +} + +func TestScanMetrics_RTT_InsufficientSamples(t *testing.T) { + m := &ScanMetrics{} + + // 少于 20 个样本 + for i := 0; i < 10; i++ { + m.RecordConnect(time.Millisecond) + } + + ratio := m.RTTRatio() + if ratio != 1.0 { + t.Errorf("样本不足时 RTTRatio = %.2f, 应该是 1.0", ratio) + } +} + +// ============================================================================= +// 并发安全测试 +// ============================================================================= + +func TestScanMetrics_ConcurrentSafety(t *testing.T) { + m := &ScanMetrics{} + var wg sync.WaitGroup + + for i := 0; i < 100; i++ { + wg.Add(4) + go func() { defer wg.Done(); m.RecordConnect(time.Millisecond) }() + go func() { defer wg.Done(); m.RecordRefused(time.Millisecond) }() + go func() { defer wg.Done(); m.RecordTimeout() }() + go func() { defer wg.Done(); m.RecordExhausted() }() + } + + wg.Wait() + + if m.Total() != 400 { + t.Errorf("并发后 Total() = %d, want 400", m.Total()) + } + + // 验证 Snapshot 不 panic + snap := m.Snapshot() + if snap.Total() != 400 { + t.Errorf("并发后 Snapshot.Total() = %d, want 400", snap.Total()) + } + + // 验证 RTTRatio 不 panic + _ = m.RTTRatio() +} diff --git a/core/service_scanner.go b/core/service_scanner.go index 82025a4..4668eb5 100644 --- a/core/service_scanner.go +++ b/core/service_scanner.go @@ -164,6 +164,10 @@ func (s *ServiceScanStrategy) performHostScan(ctx context.Context, session *comm totalAlive := 0 sawHosts := false performedLiveness := false + envProfiled := false + + // 系统能力探测（不需要网络目标） + sysProfile := ProbeSystem() for { hosts, err := iter.NextBatch(ctx, targetHostBatchSize(config)) @@ -185,6 +189,14 @@ func (s *ServiceScanStrategy) performHostScan(ctx context.Context, session *comm continue } + // 首批 alive hosts 出来后做网络探测，调整后续所有参数 + if !envProfiled { + envProfiled = true + netProfile := ProbeNetwork(ctx, hosts, session) + ep := &EnvironmentProfile{Net: *netProfile, System: sysProfile} + ep.TuneConfig(config, session) + } + s.dispatchUDPPlugins(ctx, session, hosts, info, config, ch, wg) s.scanHostBatch(ctx, session, hosts, info, pluginsToRun, isCustomMode, ch, wg) } diff --git a/web/api/scan.go b/web/api/scan.go index 195f51d..320908d 100644 --- a/web/api/scan.go +++ b/web/api/scan.go @@ -210,6 +210,11 @@ func (h *ScanHandler) runScan(req ScanRequest) { fv.DisableSave = true // Web模式不保存到文件 fv.Silent = true // 静默模式 + // 用户指定了线程数则标记为显式 + if req.ThreadNum > 0 { + fv.ThreadNumExplicit = true + } + // 构建Config和Session config := common.BuildConfigFromFlags(fv) state := common.NewState()