fix(network): 优化进程流量实时采样

2026-09-21 21:48:30 +08:00
1 parent 4457d6201a
commit b462420046
@@ -1,6 +1,23 @@
 import Darwin
 import Foundation

+private final class LockedProcessOutput: @unchecked Sendable {
+    private let lock = NSLock()
+    private var data: Data?
+
+    func store(_ data: Data) {
+        lock.lock()
+        self.data = data
+        lock.unlock()
+    }
+
+    func load() -> Data? {
+        lock.lock()
+        defer { lock.unlock() }
+        return data
+    }
+}
+
 public struct NetworkProcessUsage: Sendable, Equatable, Identifiable {
    public var id: Int32 { pid }
    public let pid: Int32
@@ -13,14 +30,20 @@ public struct NetworkProcessUsage: Sendable, Equatable, Identifiable {
 /// 各进程的网络速率。系统没有公开接口，读取 /usr/bin/nettop 的累计字节数并与上次比较。
 /// 只在网络详情打开时运行，每次约 40 ms。
 public struct NetworkProcessSampler: Sendable {
+    private static let nettopLock = NSLock()
    private var previous: [Int32: (download: UInt64, upload: UInt64)] = [:]
    private var previousTime: UInt64 = 0
+    private var lastResults: [NetworkProcessUsage] = []

    public init() {}

    public mutating func sample(limit: Int = 8) -> [NetworkProcessUsage] {
-        guard let output = Self.runNettop() else { return [] }
-        let now = clock_gettime_nsec_np(CLOCK_UPTIME_RAW)
+        sample(output: Self.runNettop(), now: clock_gettime_nsec_np(CLOCK_UPTIME_RAW), limit: limit)
+    }
+
+    /// 将 `nettop` 的累计值换算成区间速率。读取失败时保留上次结果，避免界面短暂清空。
+    mutating func sample(output: String?, now: UInt64, limit: Int = 8) -> [NetworkProcessUsage] {
+        guard let output else { return lastResults }
        let seconds = previousTime > 0 ? Double(now - previousTime) / 1_000_000_000 : 0
        let totals = Self.parse(output)

@@ -32,7 +55,8 @@ public struct NetworkProcessSampler: Sendable {
                let upload = entry.upload >= before.upload ? Double(entry.upload - before.upload) / seconds : 0
                guard download + upload > 0 else { continue }
                let meta = ProcessSampler.meta(for: pid)
-                results.append(NetworkProcessUsage(pid: pid, name: meta.bundle.map { URL(fileURLWithPath: $0).deletingPathExtension().lastPathComponent } ?? meta.name,
+                let processName = meta.name == "PID \(pid)" ? entry.name : meta.name
+                results.append(NetworkProcessUsage(pid: pid, name: meta.bundle.map { URL(fileURLWithPath: $0).deletingPathExtension().lastPathComponent } ?? processName,
                                                   appBundlePath: meta.bundle, download: download, upload: upload))
            }
        }
@@ -51,10 +75,12 @@ public struct NetworkProcessSampler: Sendable {
                merged[key] = usage
            }
        }
-        return merged.values
+        let ranked = merged.values
            .sorted { $0.download + $0.upload > $1.download + $1.upload }
            .prefix(limit)
            .map { $0 }
+        lastResults = ranked
+        return ranked
    }

    /// 解析 `nettop -P -L 1 -x -J bytes_in,bytes_out` 的 CSV：`进程名.PID,入字节,出字节,`
@@ -73,19 +99,79 @@ public struct NetworkProcessSampler: Sendable {
    }

    private static func runNettop() -> String? {
+        nettopLock.lock()
+        defer { nettopLock.unlock() }
+        guard !Task.isCancelled else { return nil }
+        return run(path: "/usr/bin/nettop", arguments: ["-P", "-L", "1", "-n", "-x", "-J", "bytes_in,bytes_out"], timeout: 5)
+    }
+
+    /// 同步运行短命令并持续排空 stdout；超时后先 TERM、再 KILL，避免 `nettop` 挂住采样循环。
+    static func run(path: String, arguments: [String], timeout: TimeInterval) -> String? {
+        guard !Task.isCancelled else { return nil }
        let process = Process()
-        process.executableURL = URL(fileURLWithPath: "/usr/bin/nettop")
-        process.arguments = ["-P", "-L", "1", "-n", "-x", "-J", "bytes_in,bytes_out"]
+        process.executableURL = URL(fileURLWithPath: path)
+        process.arguments = arguments
        let pipe = Pipe()
        process.standardOutput = pipe
        process.standardError = FileHandle.nullDevice
+
+        let output = LockedProcessOutput()
+        let drain = DispatchGroup()
+        drain.enter()
+        DispatchQueue.global(qos: .utility).async {
+            let data = pipe.fileHandleForReading.readDataToEndOfFile()
+            output.store(data)
+            drain.leave()
+        }
+
+        let terminated = DispatchSemaphore(value: 0)
+        process.terminationHandler = { _ in terminated.signal() }
        do {
            try process.run()
        } catch {
+            pipe.fileHandleForWriting.closeFile()
+            drain.wait()
+            return nil
+        }
+
+        let timeoutNanoseconds = UInt64(max(0, timeout) * 1_000_000_000)
+        let deadline = DispatchTime.now().uptimeNanoseconds &+ timeoutNanoseconds
+        var stopped = false
+        var exited = false
+        while !exited {
+            if Task.isCancelled {
+                stopped = true
+                break
+            }
+            let now = DispatchTime.now().uptimeNanoseconds
+            guard now < deadline else {
+                stopped = true
+                break
+            }
+            let slice = min(deadline - now, 50_000_000)
+            exited = terminated.wait(timeout: .now() + .nanoseconds(Int(slice))) == .success
+        }
+
+        if stopped {
+            if process.isRunning { process.terminate() }
+            exited = terminated.wait(timeout: .now() + 0.5) == .success
+            if !exited {
+                kill(process.processIdentifier, SIGKILL)
+                exited = terminated.wait(timeout: .now() + 0.5) == .success
+            }
+        }
+        guard exited else {
+            pipe.fileHandleForReading.closeFile()
            return nil
        }
-        let data = pipe.fileHandleForReading.readDataToEndOfFile()
        process.waitUntilExit()
-        return process.terminationStatus == 0 ? String(decoding: data, as: UTF8.self) : nil
+        guard drain.wait(timeout: .now() + 0.5) == .success else {
+            pipe.fileHandleForReading.closeFile()
+            return nil
+        }
+
+        guard !stopped, process.terminationStatus == 0,
+              let data = output.load() else { return nil }
+        return String(decoding: data, as: UTF8.self)
    }
 }
@@ -46,6 +46,7 @@ public final class NetworkController {
    @ObservationIgnored private let settings: AppSettings
    @ObservationIgnored private var probeTask: Task<Void, Never>?
    @ObservationIgnored private var detailTask: Task<Void, Never>?
+    @ObservationIgnored private var processSampler = NetworkProcessSampler()
    @ObservationIgnored private var probedAddress: String?
    @ObservationIgnored private var lastPublicLookup: (date: Date, localIPv4: [String])?
    @ObservationIgnored private var isPaused = false
@@ -164,7 +165,7 @@ public final class NetworkController {
        applyDetailState()
    }

-    /// 网络详情打开时每 2 秒刷新接口信息与进程流量，关闭后停止
+    /// 网络详情打开时每秒刷新进程流量、每 4 秒刷新接口信息，关闭后停止
    private func applyDetailState() {
        let shouldRun = wantsDetail && !isPaused
        guard shouldRun != (detailTask != nil) else { return }
@@ -173,26 +174,37 @@ public final class NetworkController {
        // 关闭时保留上次的进程列表，再次打开时不会先闪一下空白
        guard shouldRun else { return }
        detailTask = Task { [weak self] in
-            var sampler = NetworkProcessSampler()
            var tick = 0
+            let clock = ContinuousClock()
            while !Task.isCancelled {
-                // 接口信息每 4 秒读一次，进程流量每 2 秒
-                if tick % 2 == 0 {
+                guard let self else { return }
+                let started = clock.now
+                if tick % 4 == 0 {
                    let details = await Task.detached { NetworkDetailsReader.read() }.value
-                    guard !Task.isCancelled, let self else { return }
+                    guard !Task.isCancelled else { return }
                    if details != self.details { self.details = details }
                    if tick == 0 { self.lookUpPublicAddressesIfStale() }
                }
-                let (usage, next) = await Task.detached { [sampler] in
+                let sampler = self.processSampler
+                let samplingTask = Task.detached { [sampler] in
                    var copy = sampler
                    let usage = copy.sample()
                    return (usage, copy)
-                }.value
-                sampler = next
-                guard !Task.isCancelled, let self else { return }
+                }
+                let (usage, next) = await withTaskCancellationHandler {
+                    await samplingTask.value
+                } onCancel: {
+                    samplingTask.cancel()
+                }
+                guard !Task.isCancelled else { return }
+                self.processSampler = next
                self.processes = self.rankedProcesses(usage)
                tick += 1
-                try? await Task.sleep(for: .seconds(2))
+                do {
+                    try await Task.sleep(until: started + .seconds(1), tolerance: .milliseconds(100), clock: clock)
+                } catch {
+                    return
+                }
            }
        }
    }
@@ -156,6 +156,79 @@ struct LiveSamplerTests {
        #expect(parsed[4521]?.upload == 95_191)
    }

+    @Test func usesNettopNameWhenProcessMetadataIsUnavailable() throws {
+        var sampler = NetworkProcessSampler()
+        _ = sampler.sample(output: """
+        ,bytes_in,bytes_out,
+        kernel_task.0,1000,2000,
+        """, now: 1_000_000_000)
+
+        let usage = sampler.sample(output: """
+        ,bytes_in,bytes_out,
+        kernel_task.0,3000,5000,
+        """, now: 2_000_000_000)
+
+        let kernel = try #require(usage.first)
+        #expect(kernel.pid == 0)
+        #expect(kernel.name == "kernel_task")
+        #expect(kernel.download == 2_000)
+        #expect(kernel.upload == 3_000)
+    }
+
+    @Test func keepsLastUsageWhenNettopFails() throws {
+        var sampler = NetworkProcessSampler()
+        _ = sampler.sample(output: """
+        ,bytes_in,bytes_out,
+        kernel_task.0,1000,2000,
+        """, now: 1_000_000_000)
+        let valid = sampler.sample(output: """
+        ,bytes_in,bytes_out,
+        kernel_task.0,3000,5000,
+        """, now: 2_000_000_000)
+
+        let retained = sampler.sample(output: nil, now: 3_000_000_000)
+
+        #expect(retained == valid)
+    }
+
+    @Test func stopsNettopCommandAfterTimeout() {
+        let clock = ContinuousClock()
+        let started = clock.now
+        let output = NetworkProcessSampler.run(
+            path: "/bin/sleep",
+            arguments: ["1"],
+            timeout: 0.01
+        )
+
+        #expect(output == nil)
+        #expect(started.duration(to: clock.now) < .milliseconds(500))
+    }
+
+    @Test func cancellationStopsRunningCommand() async throws {
+        let marker = FileManager.default.temporaryDirectory.appending(path: "xstats-process-\(UUID().uuidString)")
+        defer { try? FileManager.default.removeItem(at: marker) }
+        let clock = ContinuousClock()
+        let task = Task.detached {
+            NetworkProcessSampler.run(
+                path: "/bin/sh",
+                arguments: ["-c", "touch \"$1\"; sleep 5", "xstats-test", marker.path],
+                timeout: 5
+            )
+        }
+
+        let launchDeadline = clock.now + .seconds(1)
+        while !FileManager.default.fileExists(atPath: marker.path), clock.now < launchDeadline {
+            await Task.yield()
+        }
+        try #require(FileManager.default.fileExists(atPath: marker.path))
+        let cancelled = clock.now
+        task.cancel()
+        let output = await task.value
+
+        #expect(output == nil)
+        #expect(cancelled.duration(to: clock.now) < .seconds(1))
+    }
+
    @Test func parsesCloudflareTrace() {
        let fields = PublicAddressLookup.parseTrace("fl=123\nip=203.0.113.24\nloc=CN\n")
        #expect(fields["ip"] == "203.0.113.24")