diff --git a/Tun/Punchnet/Concurrency/PeriodicWorker.swift b/Tun/Punchnet/Concurrency/PeriodicWorker.swift index 2398a21..7e83a0c 100644 --- a/Tun/Punchnet/Concurrency/PeriodicWorker.swift +++ b/Tun/Punchnet/Concurrency/PeriodicWorker.swift @@ -132,13 +132,14 @@ public actor PeriodicWorker { } private static func runFixedRate(configuration: Configuration, operation: @escaping Operation, onError: @escaping ErrorHandler) async { - var nextRun = ContinuousClock.now + let clock = ContinuousClock() + var nextRun = clock.now if !configuration.runImmediately { nextRun = nextRun.advanced(by: configuration.interval) } while !Task.isCancelled { - let now = ContinuousClock.now + let now = clock.now if nextRun > now { let delay = now.duration(to: nextRun) guard await sleep(interval: delay, tolerance: configuration.tolerance) else { @@ -155,7 +156,7 @@ public actor PeriodicWorker { return } - let afterRun = ContinuousClock.now + let afterRun = clock.now repeat { nextRun = nextRun.advanced(by: configuration.interval) } while nextRun <= afterRun diff --git a/Tun/Punchnet/Context/SDLContextActor.swift b/Tun/Punchnet/Context/SDLContextActor.swift index 8379f48..6cf2788 100644 --- a/Tun/Punchnet/Context/SDLContextActor.swift +++ b/Tun/Punchnet/Context/SDLContextActor.swift @@ -90,11 +90,11 @@ actor SDLContextActor { nonisolated private let provider: NEPacketTunnelProvider // 处理权限控制 - private var updatePolicyTask: Task? private let policyService: PolicyService + private var updatePolicyWorker: PeriodicWorker? // stunRequest任务 - private var stunRequestTask: Task? + private var stunRequestWorker: PeriodicWorker? public init(provider: NEPacketTunnelProvider, config: SDLConfiguration, rsaCipher: RSACipher) { let puncherActor = SDLPuncherActor() @@ -183,11 +183,11 @@ actor SDLContextActor { await self.arpServer.stop() await self.sessionManager.clear() - self.stunRequestTask?.cancel() - self.stunRequestTask = nil + await self.stunRequestWorker?.stop() + self.stunRequestWorker = nil - self.updatePolicyTask?.cancel() - self.updatePolicyTask = nil + await self.updatePolicyWorker?.stop() + self.updatePolicyWorker = nil await self.policyService.clear() await self.packetOutboundActor.stop() @@ -349,19 +349,27 @@ extension SDLContextActor { // 注册成功super的回调函数 private func whenRegistedSuper() async { - self.updatePolicyTask?.cancel() - - self.updatePolicyTask = Task { - do { - while true { - try await Task.sleep(for: .seconds(300)) - SDLLogger.log("[SDLContext] updatePolicyTask execute") - await self.policyService.updatePolicy(superServiceProxy: self.superServiceProxy) - } - } catch let err { + await self.updatePolicyWorker?.stop() + let policyService = self.policyService + let superServiceProxy = self.superServiceProxy + + let updatePolicyWorker = PeriodicWorker( + configuration: .init( + interval: .seconds(300), + runImmediately: false, + mode: .fixedDelay, + errorPolicy: .keepRunning(delay: .seconds(5)) + ), + operation: { + SDLLogger.log("[SDLContext] updatePolicyTask execute") + await policyService.updatePolicy(superServiceProxy: superServiceProxy) + }, + onError: { err in SDLLogger.log("[SDLContext] updatePolicyTask stop with err: \(err)") } - } + ) + self.updatePolicyWorker = updatePolicyWorker + await updatePolicyWorker.start() // 启动stun任务 await self.startStunRequestTask() @@ -519,23 +527,17 @@ extension SDLContextActor { // MARK: -- StunRequestTask private func startStunRequestTask() async { - self.stunRequestTask?.cancel() - self.stunRequestTask = nil + await self.stunRequestWorker?.stop() - // 处理心跳逻辑 - self.stunRequestTask = Task { [weak self] in - let timerStream = SDLAsyncTimerStream() - timerStream.start(interval: .seconds(8)) - - for await _ in timerStream.stream { - if Task.isCancelled { - break - } - + let stunRequestWorker = PeriodicWorker( + configuration: .init( + interval: .seconds(8), + runImmediately: true, + mode: .fixedDelay, + errorPolicy: .keepRunning(delay: .seconds(5)) + ), + operation: { [weak self] in let probeReply = try? await self?.ipv6AssistClient?.probe(requestTimeout: .seconds(3)) - if Task.isCancelled { - break - } if let v6Info = probeReply?.v6Info, let v6Address = SDLUtil.ipv6DataToString(v6Info.v6) { SDLLogger.log("[SDLContext] probe ipv6 address: \(v6Address)") @@ -543,15 +545,14 @@ extension SDLContextActor { SDLLogger.log("[SDLContext] probe ipv6 address: empty") } - if Task.isCancelled { - break - } - await self?.sendStunRequest(v6Info: probeReply?.v6Info) + }, + onError: { err in + SDLLogger.log("[SDLContext] udp stunRequestTask stop with err: \(err)") } - - SDLLogger.log("[SDLContext] udp stunRequestTask cancel") - } + ) + self.stunRequestWorker = stunRequestWorker + await stunRequestWorker.start() } private func sendStunRequest(v6Info: SDLV6Info?) async {