From b462420046023abefb44a77d516eb48dfbbcc3c1 Mon Sep 17 00:00:00 2001 From: ysicing Date: Mon, 21 Sep 2026 21:48:30 +0800 Subject: [PATCH] =?UTF-8?q?fix(network):=20=E4=BC=98=E5=8C=96=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E6=B5=81=E9=87=8F=E5=AE=9E=E6=97=B6=E9=87=87=E6=A0=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Metrics/NetworkProcessSampler.swift | 102 ++++++++++++++++-- .../XStatsUI/State/NetworkController.swift | 32 ++++-- .../Tests/MetricsTests/MetricsTests.swift | 73 +++++++++++++ 3 files changed, 189 insertions(+), 18 deletions(-) diff --git a/Packages/XStatsKit/Sources/Metrics/NetworkProcessSampler.swift b/Packages/XStatsKit/Sources/Metrics/NetworkProcessSampler.swift index 6636a7d..bd60a63 100644 --- a/Packages/XStatsKit/Sources/Metrics/NetworkProcessSampler.swift +++ b/Packages/XStatsKit/Sources/Metrics/NetworkProcessSampler.swift @@ -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) } } diff --git a/Packages/XStatsKit/Sources/XStatsUI/State/NetworkController.swift b/Packages/XStatsKit/Sources/XStatsUI/State/NetworkController.swift index 0ee41cc..b709c95 100644 --- a/Packages/XStatsKit/Sources/XStatsUI/State/NetworkController.swift +++ b/Packages/XStatsKit/Sources/XStatsUI/State/NetworkController.swift @@ -46,6 +46,7 @@ public final class NetworkController { @ObservationIgnored private let settings: AppSettings @ObservationIgnored private var probeTask: Task? @ObservationIgnored private var detailTask: Task? + @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 + } } } } diff --git a/Packages/XStatsKit/Tests/MetricsTests/MetricsTests.swift b/Packages/XStatsKit/Tests/MetricsTests/MetricsTests.swift index 2e43f63..f3d9719 100644 --- a/Packages/XStatsKit/Tests/MetricsTests/MetricsTests.swift +++ b/Packages/XStatsKit/Tests/MetricsTests/MetricsTests.swift @@ -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")