diff --git a/Tun/Context/SDLContextActor.swift b/Tun/Context/SDLContextActor.swift index ce32580..24c735f 100644 --- a/Tun/Context/SDLContextActor.swift +++ b/Tun/Context/SDLContextActor.swift @@ -55,6 +55,8 @@ actor SDLContextActor { // 本地ipv6地址信息探测 private var ipv6AssistClient: SDLIPV6AssistClient? + private let ipv6AssistEvents: AsyncStream + private let ipv6AssistContinuation: AsyncStream.Continuation private let sessionManager: SessionManager nonisolated private let arpResolver: ArpResolver @@ -82,6 +84,7 @@ actor SDLContextActor { let superService = SDLSuperService(serverEndpoint: config.serverEndpoint) let udpHoleService = SDLUDPHoleService(proberActor: proberActor) let tunNetworkManager = SDLTunNetworkManager(provider: provider) + let ipv6AssistPair = AsyncStream.makeStream(of: Optional.self, bufferingPolicy: .bufferingNewest(1)) let packetOutboundActor = PacketOutboundActor( provider: provider, config: config, @@ -123,6 +126,8 @@ actor SDLContextActor { self.packetOutboundActor = packetOutboundActor self.packetInboundActor = packetInboundActor self.tunNetworkManager = tunNetworkManager + self.ipv6AssistEvents = ipv6AssistPair.stream + self.ipv6AssistContinuation = ipv6AssistPair.continuation } public func start() async throws { @@ -231,6 +236,10 @@ actor SDLContextActor { try await arpResolver.runCleanup() } + group.addTask { + try await self.runIPv6AssistSupervisor() + } + group.addTask(priority: .high) { _ = try await readySignal.wait() try await packetOutboundActor.runPacketReader() @@ -297,6 +306,49 @@ actor SDLContextActor { } } + private func runIPv6AssistSupervisor() async throws { + do { + for await assistInfo in self.ipv6AssistEvents { + try Task.checkCancellation() + await self.stopCurrentIPv6AssistClient() + + guard let assistInfo else { + continue + } + + guard let client = SDLIPV6AssistClient(assistServerInfo: assistInfo) else { + SDLLogger.log("[SDLContext] invalid ipv6 assist config", for: .debug) + continue + } + + self.ipv6AssistClient = client + + do { + try await client.run() + } catch is CancellationError { + throw CancellationError() + } catch { + SDLLogger.log("[SDLContext] ipv6 assist client ended: \(error.localizedDescription)", for: .debug) + } + + if self.ipv6AssistClient === client { + self.ipv6AssistClient = nil + } + } + } catch is CancellationError { + await self.stopCurrentIPv6AssistClient() + throw CancellationError() + } + + await self.stopCurrentIPv6AssistClient() + } + + private func stopCurrentIPv6AssistClient() async { + let client = self.ipv6AssistClient + self.ipv6AssistClient = nil + await client?.stop() + } + private func cleanupRoot() async { await self.puncherActor.stop() await self.arpResolver.stop() @@ -319,8 +371,8 @@ actor SDLContextActor { await self.packetOutboundActor.updateRuntime(config: self.config, dataCipher: nil) await self.packetInboundActor.updateRuntime(config: self.config, dataCipher: nil) - await self.ipv6AssistClient?.stop() - self.ipv6AssistClient = nil + self.ipv6AssistContinuation.yield(nil) + await self.stopCurrentIPv6AssistClient() } deinit { @@ -379,12 +431,12 @@ extension SDLContextActor { case .welcome(let welcome): SDLLogger.log("[SDLContext] quic welcome: \(welcome)") - await self.ipv6AssistClient?.stop() - self.ipv6AssistClient = nil + await self.stopCurrentIPv6AssistClient() // 尝试创建v6地址的辅助探测器 if welcome.hasIpv6Assist { - self.ipv6AssistClient = SDLIPV6AssistClient(assistServerInfo: welcome.ipv6Assist) - await self.ipv6AssistClient?.start() + self.ipv6AssistContinuation.yield(welcome.ipv6Assist) + } else { + self.ipv6AssistContinuation.yield(nil) } // 注册 diff --git a/Tun/DNS/DNSService.swift b/Tun/DNS/DNSService.swift index ac2bdde..dc2d05b 100644 --- a/Tun/DNS/DNSService.swift +++ b/Tun/DNS/DNSService.swift @@ -84,13 +84,9 @@ actor DNSService { } group.addTask { - try await withTaskCancellationHandler { - for try await packet in dnsClient.packetFlow { - try Task.checkCancellation() - await onEvent(.packet(packet)) - } - } onCancel: { - dnsClient.stop() + for try await packet in dnsClient.packetFlow { + try Task.checkCancellation() + await onEvent(.packet(packet)) } } @@ -122,15 +118,9 @@ actor DNSService { } group.addTask { - try await withTaskCancellationHandler { - for try await packet in dnsLocalClient.packetFlow { - try Task.checkCancellation() - await onEvent(.packet(packet)) - } - } onCancel: { - Task { - await dnsLocalClient.stop() - } + for try await packet in dnsLocalClient.packetFlow { + try Task.checkCancellation() + await onEvent(.packet(packet)) } } diff --git a/Tun/Foundation/SDLThrottler.swift b/Tun/Foundation/SDLThrottler.swift deleted file mode 100644 index 4e32217..0000000 --- a/Tun/Foundation/SDLThrottler.swift +++ /dev/null @@ -1,45 +0,0 @@ -// -// SDLThrottler.swift -// Tun -// -// Created by 安礼成 on 2024/6/3. -// - -import Foundation -import Combine - -// 限流器 -actor SDLThrottler { - private var limit: Int - private var token: Int - private var cancel: AnyCancellable? - - init(limit: Int) { - self.limit = limit - self.token = limit - } - - func start() { - self.cancel?.cancel() - self.cancel = Timer.publish(every: 1.0, on: .main, in: .common).autoconnect() - .sink { _ in - Task { - self.token = self.limit - } - } - } - - func setRateLimit(limit: Int) { - self.limit = limit - } - - func getToken(num: Int) -> Bool { - if token > 0 { - self.token = self.token - num - return true - } else { - return false - } - } - -} diff --git a/Tun/NWPath/SDLPathDebounceActor.swift b/Tun/NWPath/SDLPathDebounceActor.swift deleted file mode 100644 index e050f78..0000000 --- a/Tun/NWPath/SDLPathDebounceActor.swift +++ /dev/null @@ -1,54 +0,0 @@ -// -// SDLPathDebounceActor.swift -// punchnet -// -// Created by 安礼成 on 2026/4/29. -// - -import Foundation -import Network - -actor SDLPathDebounceActor { - private var lastSnapshot: SDLPathSnapshot? - private var debounceTask: Task? - - private let delay: Duration - - public let statusStream: AsyncStream - private let statusCont: AsyncStream.Continuation - - init(delay: Duration = .seconds(2)) { - self.delay = delay - let (stream, cont) = AsyncStream.makeStream(of: NWPath.Status.self) - self.statusStream = stream - self.statusCont = cont - } - - func submit(_ snapshot: SDLPathSnapshot) { - guard snapshot != lastSnapshot else { - return - } - - self.lastSnapshot = snapshot - - self.debounceTask?.cancel() - self.debounceTask = Task { [delay] in - try? await Task.sleep(for: delay) - - guard !Task.isCancelled else { - return - } - await self.handleStableSnapshot(snapshot) - } - } - - private func handleStableSnapshot(_ snapshot: SDLPathSnapshot) async { - self.statusCont.yield(snapshot.status) - } - - func stop() { - debounceTask?.cancel() - debounceTask = nil - } - -} diff --git a/Tun/NWPath/SDLPathMonitor.swift b/Tun/NWPath/SDLPathMonitor.swift deleted file mode 100644 index e01f08e..0000000 --- a/Tun/NWPath/SDLPathMonitor.swift +++ /dev/null @@ -1,45 +0,0 @@ -// -// SDLPathMonitor.swift -// punchnet -// -// Created by 安礼成 on 2026/4/29. -// -import Foundation -import Network - -final class SDLPathMonitor { - private let monitor = NWPathMonitor() - private let queue = DispatchQueue(label: "com.sdlan.path-monitor") - private let debouncer = SDLPathDebounceActor(delay: .seconds(2)) - - private var isStarted: Bool = false - - func start() { - guard !isStarted else { - return - } - - self.isStarted = true - monitor.pathUpdateHandler = { path in - let snapshot = SDLPathSnapshot(path) - - Task { - await self.debouncer.submit(snapshot) - } - } - - monitor.start(queue: queue) - } - - func statusStream() -> AsyncStream { - return self.debouncer.statusStream - } - - func stop() { - Task { - await debouncer.stop() - } - monitor.cancel() - } - -} diff --git a/Tun/NWPath/SDLPathSnapshot.swift b/Tun/NWPath/SDLPathSnapshot.swift deleted file mode 100644 index 3b6d455..0000000 --- a/Tun/NWPath/SDLPathSnapshot.swift +++ /dev/null @@ -1,32 +0,0 @@ -// -// PathSnapshot.swift -// punchnet -// -// Created by 安礼成 on 2026/4/29. -// -import Foundation -import Network - -struct SDLPathSnapshot: Equatable, Sendable { - let status: NWPath.Status - let isExpensive: Bool - let isConstrained: Bool - let supportsIPv4: Bool - let supportsIPv6: Bool - let supportsDNS: Bool - let usesWiFi: Bool - let usesWired: Bool - let usesCellular: Bool - - init(_ path: NWPath) { - self.status = path.status - self.isExpensive = path.isExpensive - self.isConstrained = path.isConstrained - self.supportsIPv4 = path.supportsIPv4 - self.supportsIPv6 = path.supportsIPv6 - self.supportsDNS = path.supportsDNS - self.usesWiFi = path.usesInterfaceType(.wifi) - self.usesWired = path.usesInterfaceType(.wiredEthernet) - self.usesCellular = path.usesInterfaceType(.cellular) - } -} diff --git a/Tun/Network/SDLIPV6AssistClient.swift b/Tun/Network/SDLIPV6AssistClient.swift index 13d7d02..8be7911 100644 --- a/Tun/Network/SDLIPV6AssistClient.swift +++ b/Tun/Network/SDLIPV6AssistClient.swift @@ -26,7 +26,6 @@ actor SDLIPV6AssistClient { private var state: State = .idle private var connection: NWConnection? - private var receiveTask: Task? private let assistServerAddress: NWEndpoint private var packetId: UInt32 = 1 @@ -41,7 +40,7 @@ actor SDLIPV6AssistClient { self.assistServerAddress = .hostPort(host: .ipv6(address), port: NWEndpoint.Port(integerLiteral: UInt16(assistServerInfo.port))) } - func start() { + func run() async throws { guard case .idle = self.state else { return } @@ -73,6 +72,20 @@ actor SDLIPV6AssistClient { // 启动连接队列 connection.start(queue: .global()) + + defer { + self.stop() + } + + let stream = Self.makeReceiveStream(for: connection) + try await withTaskCancellationHandler { + for await data in stream { + try Task.checkCancellation() + self.handleReceivedPacket(data) + } + } onCancel: { + connection.cancel() + } } /// 接收数据的递归循环 @@ -147,8 +160,6 @@ actor SDLIPV6AssistClient { } self.state = .stopped - self.receiveTask?.cancel() - self.receiveTask = nil self.connection?.cancel() self.connection = nil self.failAllPendingRequests(error: pendingError) @@ -162,7 +173,6 @@ actor SDLIPV6AssistClient { switch state { case .ready: SDLLogger.log("[SDLIPV6AssistClient] Connection ready", for: .debug) - self.startReceiveTask(for: connection) case .failed(let error): SDLLogger.log("[SDLIPV6AssistClient] Connection failed: \(error)", for: .debug) self.stop(pendingError: error) @@ -173,24 +183,6 @@ actor SDLIPV6AssistClient { } } - 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 { - guard let self else { - break - } - await self.handleReceivedPacket(data) - } - - await self?.didFinishReceiving(for: connection) - } - } - private func handleReceivedPacket(_ data: Data) { do { let packet = try SDLV6AssistProbeReply(serializedBytes: data) @@ -203,18 +195,6 @@ actor SDLIPV6AssistClient { } } - private func didFinishReceiving(for connection: NWConnection) { - guard case .running = self.state else { - return - } - - if self.connection === connection, connection.state != .ready { - self.stop() - } else { - self.receiveTask = nil - } - } - private func nextPacketId() -> UInt32 { let packetId = self.packetId self.packetId &+= 1 diff --git a/Tun/Network/SDLNetworkMonitor.swift b/Tun/Network/SDLNetworkMonitor.swift deleted file mode 100644 index 6b4e11e..0000000 --- a/Tun/Network/SDLNetworkMonitor.swift +++ /dev/null @@ -1,74 +0,0 @@ -// -// SDLNetworkMonitor.swift -// Tun -// -// Created by 安礼成 on 2024/5/16. -// - -import Foundation -import Network -import Combine - -// 监控网络的变化 -class SDLNetworkMonitor: @unchecked Sendable { - private var monitor: NWPathMonitor - private var interfaceType: NWInterface.InterfaceType? - private let publisher = PassthroughSubject() - private var cancel: AnyCancellable? - private var isStopped = false - - public let eventStream: AsyncStream - private let eventContinuation: AsyncStream.Continuation - - enum MonitorEvent { - case changed - case unreachable - } - - init() { - self.monitor = NWPathMonitor() - (self.eventStream , self.eventContinuation) = AsyncStream.makeStream(of: MonitorEvent.self, bufferingPolicy: .unbounded) - } - - func start() { - self.monitor.pathUpdateHandler = {path in - if path.status == .satisfied { - if path.usesInterfaceType(.wifi) { - self.publisher.send(.wifi) - } else if path.usesInterfaceType(.cellular) { - self.publisher.send(.cellular) - } else if path.usesInterfaceType(.wiredEthernet) { - self.publisher.send(.wiredEthernet) - } - } else { - self.eventContinuation.yield(.unreachable) - self.interfaceType = nil - } - } - self.monitor.start(queue: DispatchQueue.global()) - - self.cancel = publisher.throttle(for: 5.0, scheduler: DispatchQueue.global(), latest: true) - .sink { type in - if self.interfaceType != nil && self.interfaceType != type { - self.eventContinuation.yield(.changed) - } - self.interfaceType = type - } - } - - func stop() { - guard !self.isStopped else { - return - } - - self.isStopped = true - self.monitor.cancel() - self.cancel?.cancel() - self.eventContinuation.finish() - } - - deinit { - self.stop() - } - -} diff --git a/Tun/Super/SDLSuperService.swift b/Tun/Super/SDLSuperService.swift index 8c812f7..9168d6c 100644 --- a/Tun/Super/SDLSuperService.swift +++ b/Tun/Super/SDLSuperService.swift @@ -99,14 +99,7 @@ final class SDLSuperSession: @unchecked Sendable { func run() async throws { do { - try await withTaskCancellationHandler { - try await self.runLoops() - } onCancel: { - Task { - await self.client.stop() - } - } - + try await self.runLoops() await self.stop() } catch { await self.stop() diff --git a/Tun/UDPHole/SDLUDPHoleService.swift b/Tun/UDPHole/SDLUDPHoleService.swift index 20b156c..333b638 100644 --- a/Tun/UDPHole/SDLUDPHoleService.swift +++ b/Tun/UDPHole/SDLUDPHoleService.swift @@ -199,26 +199,20 @@ actor SDLUDPHoleSession { await self.onEvent(.ready(localAddress)) do { - try await withTaskCancellationHandler { - try await withThrowingTaskGroup(of: Void.self) { group in - defer { - group.cancelAll() - } - - group.addTask { - try await self.readV4Loop(udpHole: udpHole) - } - - group.addTask { - await self.probeNatType(udpHole: udpHole) - } - - try await group.waitForAll() + try await withThrowingTaskGroup(of: Void.self) { group in + defer { + group.cancelAll() } - } onCancel: { - Task { - await udpHole.stop() + + group.addTask { + try await self.readV4Loop(udpHole: udpHole) } + + group.addTask { + await self.probeNatType(udpHole: udpHole) + } + + try await group.waitForAll() } } catch { await udpHole.stop()