diff --git a/Tun/Concurrency/PeriodicWorker.swift b/Tun/Concurrency/PeriodicWorker.swift deleted file mode 100644 index 7e83a0c..0000000 --- a/Tun/Concurrency/PeriodicWorker.swift +++ /dev/null @@ -1,198 +0,0 @@ -// -// PeriodicWorker.swift -// Tun -// -// Created by Codex on 2026/5/20. -// - -import Foundation - -public actor PeriodicWorker { - public typealias Operation = @Sendable () async throws -> Void - public typealias ErrorHandler = @Sendable (Error) async -> Void - - public enum Mode: Sendable { - case fixedDelay - case fixedRate - } - - public enum ErrorPolicy: Sendable { - case keepRunning(delay: Duration?) - case stop - } - - public struct Configuration: Sendable { - public var interval: Duration - public var tolerance: Duration? - public var runImmediately: Bool - public var mode: Mode - public var errorPolicy: ErrorPolicy - - public init( - interval: Duration, - tolerance: Duration? = nil, - runImmediately: Bool = true, - mode: Mode = .fixedDelay, - errorPolicy: ErrorPolicy = .keepRunning(delay: .seconds(5)) - ) { - self.interval = interval - self.tolerance = tolerance - self.runImmediately = runImmediately - self.mode = mode - self.errorPolicy = errorPolicy - } - } - - private let configuration: Configuration - private let operation: Operation - private let onError: ErrorHandler - - private var task: Task? - private var generation: UInt64 = 0 - - public init(configuration: Configuration, operation: @escaping Operation, onError: @escaping ErrorHandler = { _ in }) { - self.configuration = configuration - self.operation = operation - self.onError = onError - } - - public func start() { - guard self.task == nil else { - return - } - - self.generation &+= 1 - let currentGeneration = self.generation - let configuration = self.configuration - let operation = self.operation - let onError = self.onError - - self.task = Task { - switch configuration.mode { - case .fixedDelay: - await Self.runFixedDelay( - configuration: configuration, - operation: operation, - onError: onError - ) - case .fixedRate: - await Self.runFixedRate( - configuration: configuration, - operation: operation, - onError: onError - ) - } - self.clearIfCurrent(generation: currentGeneration) - } - } - - public func stop() async { - self.generation &+= 1 - - let task = self.task - self.task = nil - - task?.cancel() - await task?.value - } - - public var isRunning: Bool { - self.task != nil - } - - private func clearIfCurrent(generation: UInt64) { - guard self.generation == generation else { - return - } - - self.task = nil - } - - private static func runFixedDelay(configuration: Configuration, operation: @escaping Operation, onError: @escaping ErrorHandler) async { - if !configuration.runImmediately { - guard await sleep(interval: configuration.interval, tolerance: configuration.tolerance) else { - return - } - } - - while !Task.isCancelled { - let shouldContinue = await runOperation( - operation: operation, - onError: onError, - errorPolicy: configuration.errorPolicy - ) - guard shouldContinue else { - return - } - - guard await sleep(interval: configuration.interval, tolerance: configuration.tolerance) else { - return - } - } - } - - private static func runFixedRate(configuration: Configuration, operation: @escaping Operation, onError: @escaping ErrorHandler) async { - let clock = ContinuousClock() - var nextRun = clock.now - if !configuration.runImmediately { - nextRun = nextRun.advanced(by: configuration.interval) - } - - while !Task.isCancelled { - let now = clock.now - if nextRun > now { - let delay = now.duration(to: nextRun) - guard await sleep(interval: delay, tolerance: configuration.tolerance) else { - return - } - } - - let shouldContinue = await runOperation( - operation: operation, - onError: onError, - errorPolicy: configuration.errorPolicy - ) - guard shouldContinue else { - return - } - - let afterRun = clock.now - repeat { - nextRun = nextRun.advanced(by: configuration.interval) - } while nextRun <= afterRun - } - } - - private static func runOperation(operation: @escaping Operation, onError: @escaping ErrorHandler, errorPolicy: ErrorPolicy) async -> Bool { - do { - try Task.checkCancellation() - try await operation() - return true - } catch is CancellationError { - return false - } catch { - await onError(error) - - switch errorPolicy { - case .keepRunning(let delay): - if let delay { - return await sleep(interval: delay, tolerance: nil) - } - return true - case .stop: - return false - } - } - } - - private static func sleep(interval: Duration, tolerance: Duration?) async -> Bool { - do { - try await Task.sleep(for: interval, tolerance: tolerance) - return true - } catch is CancellationError { - return false - } catch { - return false - } - } -} diff --git a/Tun/DNS/DNSCloudClient.swift b/Tun/DNS/DNSCloudClient.swift index ac4539a..02be806 100644 --- a/Tun/DNS/DNSCloudClient.swift +++ b/Tun/DNS/DNSCloudClient.swift @@ -24,7 +24,6 @@ final class DNSCloudClient { private var state: State = .idle private var connection: NWConnection? - private var receiveTask: Task? private let dnsServerAddress: NWEndpoint // 用于对外输出收到的 DNS 响应包 @@ -54,7 +53,11 @@ final class DNSCloudClient { preconditionFailure("invalid DNS cloud server IP: \(ip)") } - func start() { + func run() async throws { + guard self.state == .idle else { + return + } + // 1. 配置参数：这是解决环路的关键 let parameters = NWParameters.udp // 禁止此连接走 TUN 网卡（在 NE 中 TUN 通常被归类为 .other） @@ -71,6 +74,20 @@ final class DNSCloudClient { connection.start(queue: .global()) self.connection = connection + + defer { + self.stop() + } + + let stream = Self.makeReceiveStream(for: connection) + try await withTaskCancellationHandler { + for await data in stream { + try Task.checkCancellation() + self.packetContinuation.yield(data) + } + } onCancel: { + self.stop() + } } /// 发送 DNS 查询包（由 TUN 拦截到的原始 IP 包数据） @@ -93,9 +110,6 @@ final class DNSCloudClient { self.state = .stopped - self.receiveTask?.cancel() - self.receiveTask = nil - self.connection?.cancel() self.connection = nil @@ -108,7 +122,6 @@ final class DNSCloudClient { switch state { case .ready: SDLLogger.log("[DNSClient] Connection ready", for: .debug) - self.startReceiveTask(for: connection) self.state = .running case .failed(let error): self.finishPacketContinuationIfNeed(throwing: .failed(error)) @@ -119,22 +132,6 @@ final class DNSCloudClient { } } - private func startReceiveTask(for connection: NWConnection) { - guard self.receiveTask == nil else { - return - } - - let stream = Self.makeReceiveStream(for: connection) - self.receiveTask = Task { [weak self] in - for await data in stream { - if Task.isCancelled { - break - } - self?.packetContinuation.yield(data) - } - } - } - private func finishPacketContinuationIfNeed(throwing error: DNSCloudError?) { guard !self.isPacketContinuationFinished else { return diff --git a/Tun/DNS/DNSLocalClient.swift b/Tun/DNS/DNSLocalClient.swift index 05cd0e0..0a1a5aa 100644 --- a/Tun/DNS/DNSLocalClient.swift +++ b/Tun/DNS/DNSLocalClient.swift @@ -31,9 +31,6 @@ actor DNSLocalClient { private let dnsServerEndpoint: NWEndpoint private var connection: NWConnection? - private var receiveTask: Task? - - private var cleanupTask: Task? private let timeoutInterval: TimeInterval = 3.0 nonisolated let packetFlow: AsyncThrowingStream @@ -67,7 +64,7 @@ actor DNSLocalClient { preconditionFailure("invalid public DNS server IP: \(ip)") } - func start() { + func run() async throws { guard self.state == .idle else { return } @@ -85,17 +82,43 @@ actor DNSLocalClient { } } - let cleanupTask = Task { [weak self] in - while !Task.isCancelled { - try? await Task.sleep(nanoseconds: 3 * 1_000_000_000) - await self?.performCleanup() - } - } - self.connection = connection - self.cleanupTask = cleanupTask connection.start(queue: .global()) + + defer { + self.stop() + } + + try await withTaskCancellationHandler { + try await withThrowingTaskGroup(of: Void.self) { group in + defer { + group.cancelAll() + } + + group.addTask { [weak self] in + let stream = Self.makeReceiveStream(for: connection) + for await data in stream { + try Task.checkCancellation() + guard let self else { + return + } + await self.handleResponse(data: data) + } + } + + group.addTask { [weak self] in + while !Task.isCancelled { + try await Task.sleep(for: .seconds(3)) + await self?.performCleanup() + } + } + + try await group.next() + } + } onCancel: { + connection.cancel() + } } func query(tracker: DNSTracker, dnsPayload: Data) { @@ -129,21 +152,13 @@ actor DNSLocalClient { self.state = .stopped - let receiveTask = self.receiveTask - self.receiveTask = nil - let connection = self.connection self.connection = nil - let cleanupTask = self.cleanupTask - self.cleanupTask = nil - self.pendingRequests.removeAll() self.nextTransactionID = 1 - receiveTask?.cancel() connection?.cancel() - cleanupTask?.cancel() self.finishPacketContinuationIfNeed(throwing: nil) SDLLogger.log("[SDLLocalClient] stopped") @@ -152,9 +167,7 @@ actor DNSLocalClient { private func handleConnectionStateUpdate(_ state: NWConnection.State, for conn: NWConnection) { switch state { case .ready: - if self.markConnectionReady(conn) { - self.startReceiveTask(for: conn) - } + self.markConnectionReady(conn) case .failed(let error): SDLLogger.log("[DNSLocalClient] failed with error: \(error.localizedDescription)", for: .debug) self.finishPacketContinuationIfNeed(throwing: .failed(error)) @@ -165,29 +178,6 @@ actor DNSLocalClient { } } - private func startReceiveTask(for conn: NWConnection) { - let stream = Self.makeReceiveStream(for: conn) - - let task = Task { [weak self] in - for await data in stream { - guard let self else { - break - } - await self.handleResponse(data: data) - } - } - - let shouldKeepTask = self.state != .stopped && self.isCurrentConnection(conn) - if shouldKeepTask { - self.receiveTask?.cancel() - self.receiveTask = task - } - - if !shouldKeepTask { - task.cancel() - } - } - private func finishPacketContinuationIfNeed(throwing error: DNSLocalError?) { guard !self.isPacketContinuationFinished else { return @@ -220,13 +210,12 @@ actor DNSLocalClient { self.packetContinuation.yield(packet) } - private func markConnectionReady(_ conn: NWConnection) -> Bool { + private func markConnectionReady(_ conn: NWConnection) { guard self.state != .stopped, self.isCurrentConnection(conn) else { - return false + return } self.state = .running - return true } private func isCurrentConnection(_ conn: NWConnection) -> Bool { diff --git a/Tun/DNS/DNSService.swift b/Tun/DNS/DNSService.swift index ed18068..ac2bdde 100644 --- a/Tun/DNS/DNSService.swift +++ b/Tun/DNS/DNSService.swift @@ -65,7 +65,6 @@ actor DNSService { private func runCloud() async throws { let dnsClient = DNSCloudClient(serverIP: self.serverIP, port: 15353) self.dnsClient = dnsClient - dnsClient.start() defer { dnsClient.stop() @@ -75,20 +74,33 @@ actor DNSService { } let onEvent = self.onEvent - try await withTaskCancellationHandler { - for try await packet in dnsClient.packetFlow { - try Task.checkCancellation() - await onEvent(.packet(packet)) + try await withThrowingTaskGroup(of: Void.self) { group in + defer { + group.cancelAll() } - } onCancel: { - dnsClient.stop() + + group.addTask { + try await dnsClient.run() + } + + group.addTask { + try await withTaskCancellationHandler { + for try await packet in dnsClient.packetFlow { + try Task.checkCancellation() + await onEvent(.packet(packet)) + } + } onCancel: { + dnsClient.stop() + } + } + + try await group.next() } } private func runLocal() async throws { let dnsServer = self.publicDnsServers.randomElement() ?? "223.5.5.5" let dnsLocalClient = DNSLocalClient(host: dnsServer) - await dnsLocalClient.start() self.dnsLocalClient = dnsLocalClient SDLLogger.log("[DNSService] dnsLocalClient started") @@ -100,15 +112,29 @@ actor DNSService { let onEvent = self.onEvent do { - try await withTaskCancellationHandler { - for try await packet in dnsLocalClient.packetFlow { - try Task.checkCancellation() - await onEvent(.packet(packet)) + try await withThrowingTaskGroup(of: Void.self) { group in + defer { + group.cancelAll() } - } onCancel: { - Task { - await dnsLocalClient.stop() + + group.addTask { + try await dnsLocalClient.run() } + + group.addTask { + try await withTaskCancellationHandler { + for try await packet in dnsLocalClient.packetFlow { + try Task.checkCancellation() + await onEvent(.packet(packet)) + } + } onCancel: { + Task { + await dnsLocalClient.stop() + } + } + } + + try await group.next() } await dnsLocalClient.stop() } catch { diff --git a/Tun/Super/SDLSuperClient.swift b/Tun/Super/SDLSuperClient.swift index f9cec3a..1dcbb7c 100644 --- a/Tun/Super/SDLSuperClient.swift +++ b/Tun/Super/SDLSuperClient.swift @@ -24,7 +24,7 @@ actor SDLSuperClient { private let messageCont: AsyncThrowingStream.Continuation private var isMessageContinuationFinished: Bool = false - private var readTask: Task? + private var pendingConnectionError: Error? private let connection: NWConnection private let maxBufferSize: Int @@ -64,7 +64,11 @@ actor SDLSuperClient { SDLLogger.log("[SDLSuperClient] start with tls protocol", for: .debug) } - func start() { + func run() async throws { + guard self.state == .idle else { + return + } + self.connection.stateUpdateHandler = { [weak self] state in SDLLogger.log("[SDLSuperClient] new state: \(state)", for: .debug) Task { @@ -72,6 +76,17 @@ actor SDLSuperClient { } } self.connection.start(queue: queue) + + defer { + self.stop() + } + + try await withTaskCancellationHandler { + try await self.waitUntilReady() + try await self.readLoop() + } onCancel: { + self.connection.cancel() + } } private static func makeEndpointHost(address ip: String) -> NWEndpoint.Host { @@ -89,11 +104,12 @@ actor SDLSuperClient { private func handleConnectionState(state: NWConnection.State) { switch state { case .ready: - self.startReadTask() self.state = .running case .failed(let error): + self.pendingConnectionError = SDLSuperError.connectionFailed(error) self.finishMessageContinuationIfNeed(throwing: .connectionFailed(error)) case .cancelled: + self.pendingConnectionError = SDLSuperError.connectionCancelled self.finishMessageContinuationIfNeed(throwing: .connectionCancelled) default: () @@ -113,28 +129,47 @@ actor SDLSuperClient { } } - private func startReadTask() { - self.readTask?.cancel() - - self.readTask = Task { - let frameParser = SDLSuperFrameParser(maxBufferSize: self.maxBufferSize) - do { - while true { + private func waitUntilReady() async throws { + while true { + try Task.checkCancellation() + + if case .running = self.state { + return + } + + if let pendingConnectionError { + throw pendingConnectionError + } + + try await Task.sleep(for: .milliseconds(100)) + } + } + + private func readLoop() async throws { + let frameParser = SDLSuperFrameParser(maxBufferSize: self.maxBufferSize) + do { + while true { + try Task.checkCancellation() + let data = try await Self.readOnce(connection: self.connection) + let frames = try frameParser.parseFrames(data: data) + for frame in frames { try Task.checkCancellation() - let data = try await Self.readOnce(connection: self.connection) - let frames = try frameParser.parseFrames(data: data) - for frame in frames { - try Task.checkCancellation() - if let message = SDLSuperCodec.decode(frame: frame) { - self.messageCont.yield(message) - } else { - self.finishMessageContinuationIfNeed(throwing: .decodeError("invalid message")) - } + if let message = SDLSuperCodec.decode(frame: frame) { + self.messageCont.yield(message) + } else { + throw SDLSuperError.decodeError("invalid message") } } - } catch let err { - self.finishMessageContinuationIfNeed(throwing: .internalError(err)) } + } catch is CancellationError { + throw CancellationError() + } catch let error as SDLSuperError { + self.finishMessageContinuationIfNeed(throwing: error) + throw error + } catch { + let wrappedError = SDLSuperError.internalError(error) + self.finishMessageContinuationIfNeed(throwing: wrappedError) + throw wrappedError } } @@ -192,9 +227,6 @@ actor SDLSuperClient { } self.state = .stopped - - self.readTask?.cancel() - self.readTask = nil let connection = self.connection connection.stateUpdateHandler = nil diff --git a/Tun/Super/SDLSuperService.swift b/Tun/Super/SDLSuperService.swift index 78406ba..8c812f7 100644 --- a/Tun/Super/SDLSuperService.swift +++ b/Tun/Super/SDLSuperService.swift @@ -98,8 +98,6 @@ final class SDLSuperSession: @unchecked Sendable { } func run() async throws { - await self.client.start() - do { try await withTaskCancellationHandler { try await self.runLoops() @@ -125,9 +123,6 @@ final class SDLSuperSession: @unchecked Sendable { } private func runLoops() async throws { - try await Task.sleep(for: .seconds(0.5)) - try Task.checkCancellation() - SDLLogger.log("[SDLSuperSession] start super client: \(self.serverEndpoint.ip)") try await withThrowingTaskGroup(of: Void.self) { group in @@ -135,6 +130,10 @@ final class SDLSuperSession: @unchecked Sendable { group.cancelAll() } + group.addTask { + try await self.client.run() + } + group.addTask { try await self.readLoop() } diff --git a/Tun/Super/SDLSuperServiceProxy.swift b/Tun/Super/SDLSuperServiceProxy.swift deleted file mode 100644 index aeb9360..0000000 --- a/Tun/Super/SDLSuperServiceProxy.swift +++ /dev/null @@ -1,36 +0,0 @@ -// -// SDLSuperServiceProxy.swift -// punchnet -// -// Created by 安礼成 on 2026/5/21. -// -import Foundation - -actor SDLSuperServiceProxy { - private var superService: SDLSuperService? - private var generation: UInt64 = 0 - - func replace(_ superService: SDLSuperService?) async { - self.generation &+= 1 - - let oldSuperService = self.superService - self.superService = superService - - if oldSuperService !== superService { - await oldSuperService?.stop() - } - } - - func stop() async { - self.generation &+= 1 - - let superService = self.superService - self.superService = nil - - await superService?.stop() - } - - func send(type: SDLPacketType, data: Data) async { - await self.superService?.send(type: type, data: data) - } -} diff --git a/Tun/UDPHole/SDLUDPHoleServiceProxy.swift b/Tun/UDPHole/SDLUDPHoleServiceProxy.swift deleted file mode 100644 index e4c7aa7..0000000 --- a/Tun/UDPHole/SDLUDPHoleServiceProxy.swift +++ /dev/null @@ -1,54 +0,0 @@ -// -// SDLUDPHoleServiceProxy.swift -// punchnet -// -// Created by 安礼成 on 2026/5/21. -// -import Foundation -import NIOCore - -actor SDLUDPHoleServiceProxy { - typealias ControlEventHandler = @Sendable (SDLUDPHoleService.Event) async -> Void - - private var udpHoleService: SDLUDPHoleService? - private var generation: UInt64 = 0 - - func makeEventHandler(onControlEvent: @escaping ControlEventHandler) -> SDLUDPHoleService.EventHandler { - self.generation &+= 1 - let generation = self.generation - - return { [weak self] event in - await self?.handleEvent(event, generation: generation, onControlEvent: onControlEvent) - } - } - - func replace(_ udpHoleService: SDLUDPHoleService?) async { - let oldUDPHoleService = self.udpHoleService - self.udpHoleService = udpHoleService - - if oldUDPHoleService !== udpHoleService { - await oldUDPHoleService?.stop() - } - } - - func stop() async { - self.generation &+= 1 - - let udpHoleService = self.udpHoleService - self.udpHoleService = nil - - await udpHoleService?.stop() - } - - func send(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async { - await self.udpHoleService?.send(type: type, data: data, remoteAddress: remoteAddress) - } - - private func handleEvent(_ event: SDLUDPHoleService.Event, generation: UInt64, onControlEvent: ControlEventHandler) async { - guard generation == self.generation else { - return - } - - await onControlEvent(event) - } -}