archive/fscan
1
0
Fork 0
mirror of https://github.com/shadow1ng/fscan.git synced 2026-09-22 11:20:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: align atomic counters on arm

Browse Source
This commit is contained in:
ZacharyZcR
2026-06-01 03:03:40 +08:00
parent 3e4e2db722
commit 8b558b4f12
8 changed files with 161 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/progress_manager.go
+28 -31
View File
@@ -28,8 +28,8 @@ ProgressManager.go - 固定底部进度条管理器
type ProgressManager struct {
mu sync.RWMutex
enabled bool
total int64
current int64
total atomic.Int64
current atomic.Int64
description string
startTime time.Time
isActive bool
@@ -117,8 +117,8 @@ func (pm *ProgressManager) InitProgress(total int64, description string) {
pm.mu.Lock()
defer pm.mu.Unlock()
pm.total = total
pm.current = 0
pm.total.Store(total)
pm.current.Store(0)
pm.description = description
pm.startTime = time.Now()
pm.isActive = true
@@ -144,9 +144,9 @@ func (pm *ProgressManager) UpdateProgress(increment int64) {
}
// 原子累加，避免高并发下的锁竞争
newCurrent := atomic.AddInt64(&pm.current, increment)
if newCurrent > pm.total {
atomic.StoreInt64(&pm.current, pm.total)
newCurrent := pm.current.Add(increment)
if newCurrent > pm.total.Load() {
pm.current.Store(pm.total.Load())
}
// 节流渲染：距上次渲染不足 50ms 则跳过
@@ -178,7 +178,7 @@ func (pm *ProgressManager) FinishProgress() {
pm.mu.Lock()
defer pm.mu.Unlock()
atomic.StoreInt64(&pm.current, pm.total)
pm.current.Store(pm.total.Load())
pm.renderProgress()
// 停止活跃指示器
@@ -219,7 +219,7 @@ func (pm *ProgressManager) generateProgressBar() string {
// 获取发包统计
packetInfo := pm.getPacketInfo()
if pm.total == 0 {
if pm.total.Load() == 0 {
spinner := pm.getActivityIndicator()
base := fmt.Sprintf("%s %s %s", pm.description, spinner, i18n.GetText("progress_waiting"))
if packetInfo != "" {
@@ -228,9 +228,9 @@ func (pm *ProgressManager) generateProgressBar() string {
return base
}
percentage := float64(atomic.LoadInt64(&pm.current)) / float64(pm.total) * 100
percentage := float64(pm.current.Load()) / float64(pm.total.Load()) * 100
elapsed := time.Since(pm.startTime)
current := atomic.LoadInt64(&pm.current)
current := pm.current.Load()
// 计算速度
speed := float64(current) / elapsed.Seconds()
@@ -241,8 +241,8 @@ func (pm *ProgressManager) generateProgressBar() string {
// 计算预估剩余时间
var eta string
if current > 0 && current < pm.total {
totalTime := elapsed * time.Duration(pm.total) / time.Duration(current)
if current > 0 && current < pm.total.Load() {
totalTime := elapsed * time.Duration(pm.total.Load()) / time.Duration(current)
remaining := totalTime - elapsed
if remaining > 0 {
eta = fmt.Sprintf(" ETA:%s", formatDuration(remaining))
@@ -254,7 +254,7 @@ func (pm *ProgressManager) generateProgressBar() string {
// 计算固定部分的宽度
fixedPart := fmt.Sprintf("%s %s %5.1f%% [] (%d/%d)%s%s %s",
pm.description, spinner, percentage, current, pm.total, speedStr, eta, packetInfo)
pm.description, spinner, percentage, current, pm.total.Load(), speedStr, eta, packetInfo)
fixedWidth := displayWidth(fixedPart)
// 计算进度条槽位可用宽度（预留2字符余量）
@@ -281,7 +281,7 @@ func (pm *ProgressManager) generateProgressBar() string {
// 构建最终进度条
result := fmt.Sprintf("%s %s %5.1f%% %s (%d/%d)%s%s",
pm.description, spinner, percentage, bar, current, pm.total, speedStr, eta)
pm.description, spinner, percentage, bar, current, pm.total.Load(), speedStr, eta)
if packetInfo != "" {
result += " " + packetInfo
@@ -323,10 +323,10 @@ func (pm *ProgressManager) showCompletionInfo() {
durationMsg := i18n.GetText("progress_duration")
if pm.noColor {
fmt.Printf("[%s] %s %d/%d (%s: %s)\n",
doneMsg, completionMsg, pm.total, pm.total, durationMsg, formatDuration(elapsed))
doneMsg, completionMsg, pm.total.Load(), pm.total.Load(), durationMsg, formatDuration(elapsed))
} else {
fmt.Printf("%s[%s] %s %d/%d%s %s(%s: %s)%s\n",
AnsiGreen, doneMsg, completionMsg, pm.total, pm.total, AnsiReset,
AnsiGreen, doneMsg, completionMsg, pm.total.Load(), pm.total.Load(), AnsiReset,
AnsiGray, durationMsg, formatDuration(elapsed), AnsiReset)
}
}
@@ -478,10 +478,10 @@ func (pm *ProgressManager) GetPercent() float64 {
pm.mu.RLock()
defer pm.mu.RUnlock()
if !pm.isActive || pm.total == 0 {
if !pm.isActive || pm.total.Load() == 0 {
return 0
}
return float64(atomic.LoadInt64(&pm.current)) / float64(pm.total) * 100
return float64(pm.current.Load()) / float64(pm.total.Load()) * 100
}
// =============================================================================
@@ -522,8 +522,8 @@ func (pm *ProgressManager) renderProgressUnsafe() {
// 计算当前百分比（避免除零）
currentPercent := 0
if pm.total > 0 {
currentPercent = int((atomic.LoadInt64(&pm.current) * 100) / pm.total)
if pm.total.Load() > 0 {
currentPercent = int((pm.current.Load() * 100) / pm.total.Load())
}
// 只在百分比变化时更新，减少不必要的渲染
@@ -642,8 +642,8 @@ ConcurrencyMonitor - 并发监控器
// ConcurrencyMonitor 并发监控器
type ConcurrencyMonitor struct {
// 主扫描器层级
activePluginTasks int64 // 当前活跃的插件任务数
totalPluginTasks int64 // 总插件任务数
activePluginTasks atomic.Int64 // 当前活跃的插件任务数
totalPluginTasks atomic.Int64 // 总插件任务数
// 插件内连接层级已移除 - 原代码为死代码，无任何调用者
}
@@ -658,10 +658,7 @@ var (
// GetConcurrencyMonitor 获取全局并发监控器
func GetConcurrencyMonitor() *ConcurrencyMonitor {
concurrencyMutex.Do(func() {
globalConcurrencyMonitor = &ConcurrencyMonitor{
activePluginTasks: 0,
totalPluginTasks: 0,
}
globalConcurrencyMonitor = &ConcurrencyMonitor{}
})
return globalConcurrencyMonitor
}
@@ -672,18 +669,18 @@ func GetConcurrencyMonitor() *ConcurrencyMonitor {
// StartPluginTask 开始插件任务
func (m *ConcurrencyMonitor) StartPluginTask() {
atomic.AddInt64(&m.activePluginTasks, 1)
atomic.AddInt64(&m.totalPluginTasks, 1)
m.activePluginTasks.Add(1)
m.totalPluginTasks.Add(1)
}
// FinishPluginTask 完成插件任务
func (m *ConcurrencyMonitor) FinishPluginTask() {
atomic.AddInt64(&m.activePluginTasks, -1)
m.activePluginTasks.Add(-1)
}
// GetPluginTaskStats 获取插件任务统计
func (m *ConcurrencyMonitor) GetPluginTaskStats() (active int64, total int64) {
return atomic.LoadInt64(&m.activePluginTasks), atomic.LoadInt64(&m.totalPluginTasks)
return m.activePluginTasks.Load(), m.totalPluginTasks.Load()
}
// =============================================================================
common/proxy/httpdialer.go
+7 -14
View File
@@ -7,7 +7,6 @@ import (
"fmt"
"net"
"net/http"
"sync/atomic"
"time"
)
@@ -24,33 +23,27 @@ func (h *httpDialer) Dial(network, address string) (net.Conn, error) {
func (h *httpDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
start := time.Now()
atomic.AddInt64(&h.stats.TotalConnections, 1)
h.stats.addTotal(1)
// 连接到HTTP代理服务器
proxyConn, err := h.baseDial.DialContext(ctx, NetworkTCP, h.config.Address)
if err != nil {
atomic.AddInt64(&h.stats.FailedConnections, 1)
h.stats.mu.Lock()
h.stats.LastError = err.Error()
h.stats.mu.Unlock()
h.stats.addFailed(1)
h.stats.setLastError(err.Error())
return nil, NewProxyError(ErrTypeConnection, ErrMsgHTTPConnFailed, ErrCodeHTTPConnFailed, err)
}
// 发送CONNECT请求
if err := h.sendConnectRequest(proxyConn, address); err != nil {
_ = proxyConn.Close() // 错误处理路径，Close错误可忽略
atomic.AddInt64(&h.stats.FailedConnections, 1)
h.stats.mu.Lock()
h.stats.LastError = err.Error()
h.stats.mu.Unlock()
h.stats.addFailed(1)
h.stats.setLastError(err.Error())
return nil, err
}
duration := time.Since(start)
h.stats.mu.Lock()
h.stats.LastConnectTime = start
h.stats.mu.Unlock()
atomic.AddInt64(&h.stats.ActiveConnections, 1)
h.stats.setLastConnectTime(start)
h.stats.addActive(1)
h.updateAverageConnectTime(duration)
return &trackedConn{
common/proxy/manager.go
+13 -36
View File
@@ -5,7 +5,6 @@ import (
"fmt"
"net"
"sync"
"sync/atomic"
"time"
"golang.org/x/net/proxy"
@@ -127,19 +126,7 @@ func (m *manager) Stats() *ProxyStats {
m.mu.RLock()
defer m.mu.RUnlock()
m.stats.mu.Lock()
defer m.stats.mu.Unlock()
return &ProxyStats{
TotalConnections: atomic.LoadInt64(&m.stats.TotalConnections),
ActiveConnections: atomic.LoadInt64(&m.stats.ActiveConnections),
FailedConnections: atomic.LoadInt64(&m.stats.FailedConnections),
AverageConnectTime: m.stats.AverageConnectTime,
LastConnectTime: m.stats.LastConnectTime,
LastError: m.stats.LastError,
ProxyType: m.stats.ProxyType,
ProxyAddress: m.stats.ProxyAddress,
}
return m.stats.snapshot()
}
// createDirectDialer 创建直连拨号器
@@ -246,7 +233,7 @@ func (d *directDialer) Dial(network, address string) (net.Conn, error) {
func (d *directDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
start := time.Now()
atomic.AddInt64(&d.stats.TotalConnections, 1)
d.stats.addTotal(1)
dialer := &net.Dialer{
Timeout: d.timeout,
@@ -263,19 +250,15 @@ func (d *directDialer) DialContext(ctx context.Context, network, address string)
duration := time.Since(start)
d.stats.mu.Lock()
d.stats.LastConnectTime = start
d.stats.mu.Unlock()
d.stats.setLastConnectTime(start)
if err != nil {
atomic.AddInt64(&d.stats.FailedConnections, 1)
d.stats.mu.Lock()
d.stats.LastError = err.Error()
d.stats.mu.Unlock()
d.stats.addFailed(1)
d.stats.setLastError(err.Error())
return nil, NewProxyError(ErrTypeConnection, ErrMsgDirectConnFailed, ErrCodeDirectConnFailed, err)
}
atomic.AddInt64(&d.stats.ActiveConnections, 1)
d.stats.addActive(1)
d.updateAverageConnectTime(duration)
return &trackedConn{
@@ -297,7 +280,7 @@ func (s *socks5Dialer) Dial(network, address string) (net.Conn, error) {
func (s *socks5Dialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
start := time.Now()
atomic.AddInt64(&s.stats.TotalConnections, 1)
s.stats.addTotal(1)
// 创建一个带超时的上下文
dialCtx, cancel := context.WithTimeout(ctx, s.config.Timeout)
@@ -325,27 +308,21 @@ func (s *socks5Dialer) DialContext(ctx context.Context, network, address string)
select {
case <-dialCtx.Done():
atomic.AddInt64(&s.stats.FailedConnections, 1)
s.stats.mu.Lock()
s.stats.LastError = dialCtx.Err().Error()
s.stats.mu.Unlock()
s.stats.addFailed(1)
s.stats.setLastError(dialCtx.Err().Error())
return nil, NewProxyError(ErrTypeTimeout, ErrMsgSOCKS5ConnTimeout, ErrCodeSOCKS5ConnTimeout, dialCtx.Err())
case result := <-connChan:
duration := time.Since(start)
s.stats.mu.Lock()
s.stats.LastConnectTime = start
s.stats.mu.Unlock()
s.stats.setLastConnectTime(start)
if result.err != nil {
atomic.AddInt64(&s.stats.FailedConnections, 1)
s.stats.mu.Lock()
s.stats.LastError = result.err.Error()
s.stats.mu.Unlock()
s.stats.addFailed(1)
s.stats.setLastError(result.err.Error())
return nil, NewProxyError(ErrTypeConnection, ErrMsgSOCKS5ConnFailed, ErrCodeSOCKS5ConnFailed, result.err)
}
atomic.AddInt64(&s.stats.ActiveConnections, 1)
s.stats.addActive(1)
s.updateAverageConnectTime(duration)
return &trackedConn{
common/proxy/tlsdialer.go
+8 -10
View File
@@ -49,10 +49,8 @@ func (t *tlsDialerWrapper) DialTLSContext(ctx context.Context, network, address
// 进行TLS握手
if err := tlsConn.Handshake(); err != nil {
_ = tcpConn.Close() // TLS握手失败，Close错误可忽略
atomic.AddInt64(&t.stats.FailedConnections, 1)
t.stats.mu.Lock()
t.stats.LastError = err.Error()
t.stats.mu.Unlock()
t.stats.addFailed(1)
t.stats.setLastError(err.Error())
return nil, NewProxyError(ErrTypeConnection, ErrMsgTLSHandshakeFailed, ErrCodeTLSHandshakeFailed, err)
}
@@ -84,16 +82,16 @@ func (t *tlsDialerWrapper) updateAverageConnectTime(duration time.Duration) {
// trackedConn 带统计的连接
type trackedConn struct {
bytesSent atomic.Int64
bytesRecv atomic.Int64
net.Conn
stats *ProxyStats
bytesSent int64
bytesRecv int64
stats *ProxyStats
}
func (tc *trackedConn) Read(b []byte) (n int, err error) {
n, err = tc.Conn.Read(b)
if n > 0 {
atomic.AddInt64(&tc.bytesRecv, int64(n))
tc.bytesRecv.Add(int64(n))
}
return n, err
}
@@ -101,13 +99,13 @@ func (tc *trackedConn) Read(b []byte) (n int, err error) {
func (tc *trackedConn) Write(b []byte) (n int, err error) {
n, err = tc.Conn.Write(b)
if n > 0 {
atomic.AddInt64(&tc.bytesSent, int64(n))
tc.bytesSent.Add(int64(n))
}
return n, err
}
func (tc *trackedConn) Close() error {
atomic.AddInt64(&tc.stats.ActiveConnections, -1)
tc.stats.addActive(-1)
return tc.Conn.Close()
}
common/proxy/types.go
+49 -3
View File
@@ -96,9 +96,9 @@ type ProxyManager interface {
//
//nolint:revive // 保持与现有代码的向后兼容性
type ProxyStats struct {
TotalConnections int64 `json:"total_connections"`
ActiveConnections int64 `json:"active_connections"`
FailedConnections int64 `json:"failed_connections"`
TotalConnections int64 `json:"total_connections"`
ActiveConnections int64 `json:"active_connections"`
FailedConnections int64 `json:"failed_connections"`
mu sync.Mutex `json:"-"`
AverageConnectTime time.Duration `json:"average_connect_time"`
LastConnectTime time.Time `json:"last_connect_time"`
@@ -107,6 +107,52 @@ type ProxyStats struct {
ProxyAddress string `json:"proxy_address"`
}
func (s *ProxyStats) addTotal(delta int64) {
s.mu.Lock()
s.TotalConnections += delta
s.mu.Unlock()
}
func (s *ProxyStats) addActive(delta int64) {
s.mu.Lock()
s.ActiveConnections += delta
s.mu.Unlock()
}
func (s *ProxyStats) addFailed(delta int64) {
s.mu.Lock()
s.FailedConnections += delta
s.mu.Unlock()
}
func (s *ProxyStats) setLastConnectTime(t time.Time) {
s.mu.Lock()
s.LastConnectTime = t
s.mu.Unlock()
}
func (s *ProxyStats) setLastError(err string) {
s.mu.Lock()
s.LastError = err
s.mu.Unlock()
}
func (s *ProxyStats) snapshot() *ProxyStats {
s.mu.Lock()
defer s.mu.Unlock()
return &ProxyStats{
TotalConnections: s.TotalConnections,
ActiveConnections: s.ActiveConnections,
FailedConnections: s.FailedConnections,
AverageConnectTime: s.AverageConnectTime,
LastConnectTime: s.LastConnectTime,
LastError: s.LastError,
ProxyType: s.ProxyType,
ProxyAddress: s.ProxyAddress,
}
}
// ProxyError 代理错误类型
//
//nolint:revive // 保持与现有代码的向后兼容性
common/state.go
+49 -49
View File
@@ -23,17 +23,17 @@ state.go - 运行时状态管理
// State 扫描器运行时状态 - 线程安全
type State struct {
// 计数器 - 原子操作
packetCount int64
tcpPacketCount int64
tcpSuccessPacketCount int64
tcpFailedPacketCount int64
udpPacketCount int64
httpPacketCount int64
resourceExhaustedCount int64
packetCount atomic.Int64
tcpPacketCount atomic.Int64
tcpSuccessPacketCount atomic.Int64
tcpFailedPacketCount atomic.Int64
udpPacketCount atomic.Int64
httpPacketCount atomic.Int64
resourceExhaustedCount atomic.Int64
// 任务计数
end int64
num int64
end atomic.Int64
num atomic.Int64
// 时间
startTime time.Time
@@ -71,38 +71,38 @@ func NewState() *State {
// IncrementPacketCount 增加总包计数
func (s *State) IncrementPacketCount() int64 {
return atomic.AddInt64(&s.packetCount, 1)
return s.packetCount.Add(1)
}
// IncrementTCPSuccessPacketCount 增加TCP成功连接包计数
func (s *State) IncrementTCPSuccessPacketCount() int64 {
atomic.AddInt64(&s.tcpSuccessPacketCount, 1)
atomic.AddInt64(&s.tcpPacketCount, 1)
return atomic.AddInt64(&s.packetCount, 1)
s.tcpSuccessPacketCount.Add(1)
s.tcpPacketCount.Add(1)
return s.packetCount.Add(1)
}
// IncrementTCPFailedPacketCount 增加TCP失败连接包计数
func (s *State) IncrementTCPFailedPacketCount() int64 {
atomic.AddInt64(&s.tcpFailedPacketCount, 1)
atomic.AddInt64(&s.tcpPacketCount, 1)
return atomic.AddInt64(&s.packetCount, 1)
s.tcpFailedPacketCount.Add(1)
s.tcpPacketCount.Add(1)
return s.packetCount.Add(1)
}
// IncrementUDPPacketCount 增加UDP包计数
func (s *State) IncrementUDPPacketCount() int64 {
atomic.AddInt64(&s.udpPacketCount, 1)
return atomic.AddInt64(&s.packetCount, 1)
s.udpPacketCount.Add(1)
return s.packetCount.Add(1)
}
// IncrementHTTPPacketCount 增加HTTP包计数
func (s *State) IncrementHTTPPacketCount() int64 {
atomic.AddInt64(&s.httpPacketCount, 1)
return atomic.AddInt64(&s.packetCount, 1)
s.httpPacketCount.Add(1)
return s.packetCount.Add(1)
}
// IncrementResourceExhaustedCount 增加资源耗尽错误计数
func (s *State) IncrementResourceExhaustedCount() {
atomic.AddInt64(&s.resourceExhaustedCount, 1)
s.resourceExhaustedCount.Add(1)
}
// =============================================================================
@@ -111,48 +111,48 @@ func (s *State) IncrementResourceExhaustedCount() {
// GetPacketCount 获取总包计数
func (s *State) GetPacketCount() int64 {
return atomic.LoadInt64(&s.packetCount)
return s.packetCount.Load()
}
// GetTCPPacketCount 获取TCP包计数
func (s *State) GetTCPPacketCount() int64 {
return atomic.LoadInt64(&s.tcpPacketCount)
return s.tcpPacketCount.Load()
}
// GetTCPSuccessPacketCount 获取TCP成功连接包计数
func (s *State) GetTCPSuccessPacketCount() int64 {
return atomic.LoadInt64(&s.tcpSuccessPacketCount)
return s.tcpSuccessPacketCount.Load()
}
// GetTCPFailedPacketCount 获取TCP失败连接包计数
func (s *State) GetTCPFailedPacketCount() int64 {
return atomic.LoadInt64(&s.tcpFailedPacketCount)
return s.tcpFailedPacketCount.Load()
}
// GetUDPPacketCount 获取UDP包计数
func (s *State) GetUDPPacketCount() int64 {
return atomic.LoadInt64(&s.udpPacketCount)
return s.udpPacketCount.Load()
}
// GetHTTPPacketCount 获取HTTP包计数
func (s *State) GetHTTPPacketCount() int64 {
return atomic.LoadInt64(&s.httpPacketCount)
return s.httpPacketCount.Load()
}
// GetResourceExhaustedCount 获取资源耗尽错误计数
func (s *State) GetResourceExhaustedCount() int64 {
return atomic.LoadInt64(&s.resourceExhaustedCount)
return s.resourceExhaustedCount.Load()
}
// ResetPacketCounters 重置所有包计数器
func (s *State) ResetPacketCounters() {
atomic.StoreInt64(&s.packetCount, 0)
atomic.StoreInt64(&s.tcpPacketCount, 0)
atomic.StoreInt64(&s.tcpSuccessPacketCount, 0)
atomic.StoreInt64(&s.tcpFailedPacketCount, 0)
atomic.StoreInt64(&s.udpPacketCount, 0)
atomic.StoreInt64(&s.httpPacketCount, 0)
atomic.StoreInt64(&s.resourceExhaustedCount, 0)
s.packetCount.Store(0)
s.tcpPacketCount.Store(0)
s.tcpSuccessPacketCount.Store(0)
s.tcpFailedPacketCount.Store(0)
s.udpPacketCount.Store(0)
s.httpPacketCount.Store(0)
s.resourceExhaustedCount.Store(0)
}
// =============================================================================
@@ -161,32 +161,32 @@ func (s *State) ResetPacketCounters() {
// GetEnd 获取结束计数
func (s *State) GetEnd() int64 {
return atomic.LoadInt64(&s.end)
return s.end.Load()
}
// GetNum 获取数量计数
func (s *State) GetNum() int64 {
return atomic.LoadInt64(&s.num)
return s.num.Load()
}
// IncrementEnd 增加结束计数
func (s *State) IncrementEnd() int64 {
return atomic.AddInt64(&s.end, 1)
return s.end.Add(1)
}
// IncrementNum 增加数量计数
func (s *State) IncrementNum() int64 {
return atomic.AddInt64(&s.num, 1)
return s.num.Add(1)
}
// SetEnd 设置结束计数
func (s *State) SetEnd(val int64) {
atomic.StoreInt64(&s.end, val)
s.end.Store(val)
}
// SetNum 设置数量计数
func (s *State) SetNum(val int64) {
atomic.StoreInt64(&s.num, val)
s.num.Store(val)
}
// =============================================================================
@@ -271,10 +271,10 @@ type PerfStatsData struct {
func (s *State) GetPerfStats() PerfStatsData {
duration := time.Since(s.startTime)
durationMs := duration.Milliseconds()
totalPackets := atomic.LoadInt64(&s.packetCount)
tcpSuccess := atomic.LoadInt64(&s.tcpSuccessPacketCount)
tcpFailed := atomic.LoadInt64(&s.tcpFailedPacketCount)
tcpTotal := atomic.LoadInt64(&s.tcpPacketCount)
totalPackets := s.packetCount.Load()
tcpSuccess := s.tcpSuccessPacketCount.Load()
tcpFailed := s.tcpFailedPacketCount.Load()
tcpTotal := s.tcpPacketCount.Load()
var pps float64
if durationMs > 0 {
@@ -291,13 +291,13 @@ func (s *State) GetPerfStats() PerfStatsData {
TCPPackets: tcpTotal,
TCPSuccess: tcpSuccess,
TCPFailed: tcpFailed,
UDPPackets: atomic.LoadInt64(&s.udpPacketCount),
HTTPPackets: atomic.LoadInt64(&s.httpPacketCount),
ResourceExhausted: atomic.LoadInt64(&s.resourceExhaustedCount),
UDPPackets: s.udpPacketCount.Load(),
HTTPPackets: s.httpPacketCount.Load(),
ResourceExhausted: s.resourceExhaustedCount.Load(),
ScanDurationMs: durationMs,
PacketsPerSecond: pps,
SuccessRate: successRate,
TargetsScanned: atomic.LoadInt64(&s.num),
TargetsScanned: s.num.Load(),
}
}
core/adaptive_pool.go
+3 -3
View File
@@ -24,7 +24,7 @@ type AdaptivePool struct {
// 监控参数
checkInterval time.Duration
lastCheckNano int64 // 原子, UnixNano
lastCheckNano atomic.Int64 // UnixNano
lastExhaustedCount int64
lastPacketCount int64
@@ -70,12 +70,12 @@ func (ap *AdaptivePool) Invoke(task interface{}) error {
// maybeAdjust 检查并可能调整线程池大小
// 使用原子 CAS 进行时间检查，99%+ 的调用零锁开销
func (ap *AdaptivePool) maybeAdjust() {
lastCheck := atomic.LoadInt64(&ap.lastCheckNano)
lastCheck := ap.lastCheckNano.Load()
now := time.Now().UnixNano()
if now-lastCheck < int64(ap.checkInterval) {
return
}
if !atomic.CompareAndSwapInt64(&ap.lastCheckNano, lastCheck, now) {
if !ap.lastCheckNano.CompareAndSwap(lastCheck, now) {
return // 其他 goroutine 已在检查
}
core/port_scan.go
+4 -4
View File
@@ -210,7 +210,7 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout
// 初始化并发控制
to := time.Duration(timeout) * time.Second
adaptiveTO := NewAdaptiveTimeout(to)
var count int64
var count atomic.Int64
collector := newResultCollector(stream)
failedCollector := &failedPortCollector{}
var wg sync.WaitGroup
@@ -258,7 +258,7 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout
common.FinishProgressBar()
}
session.LogInfo(i18n.Tr("port_scan_complete", count))
session.LogInfo(i18n.Tr("port_scan_complete", count.Load()))
// 检查扫描失败率，如果过高则警告用户
resourceErrors := state.GetResourceExhaustedCount()
@@ -453,7 +453,7 @@ func buildServiceLogMessage(addr string, serviceInfo *ServiceInfo, isWeb bool) s
}
// scanSinglePort 扫描单个端口并进行服务识别（重构后的简洁版本）
func scanSinglePort(ctx context.Context, host string, port int, addr string, adaptiveTO *AdaptiveTimeout, count *int64, collector *resultCollector, failedCollector *failedPortCollector, session *common.ScanSession) {
func scanSinglePort(ctx context.Context, host string, port int, addr string, adaptiveTO *AdaptiveTimeout, count *atomic.Int64, collector *resultCollector, failedCollector *failedPortCollector, session *common.ScanSession) {
config := session.Config
timeout := adaptiveTO.Timeout()
// 步骤1：建立连接
@@ -486,7 +486,7 @@ func scanSinglePort(ctx context.Context, host string, port int, addr string, ada
}
// 步骤2：记录开放端口
atomic.AddInt64(count, 1)
count.Add(1)
collector.Add(addr)
saveOpenPort(session, host, port)