diff --git a/Tun/Context/SDLContextActor.swift b/Tun/Context/SDLContextActor.swift index 34e9f48..4ca0722 100644 --- a/Tun/Context/SDLContextActor.swift +++ b/Tun/Context/SDLContextActor.swift @@ -287,27 +287,19 @@ actor SDLContextActor { } group.addTask { - try await Self.runRestarting(name: "udpHoleService") { - try await udpHoleService.run() - } + try await udpHoleService.run() } group.addTask { - try await Self.runRestarting(name: "udpHoleV6Service") { - try await udpHoleV6Service.run() - } + try await udpHoleV6Service.run() } group.addTask { - try await Self.runRestarting(name: "dnsCloudService") { - try await dnsCloudService.run() - } + try await dnsCloudService.run() } group.addTask { - try await Self.runRestarting(name: "dnsLocalService") { - try await dnsLocalService.run() - } + try await dnsLocalService.run() } group.addTask { @@ -617,27 +609,6 @@ extension SDLContextActor { } extension SDLContextActor { - - private static func runRestarting( - name: String, - retryDelay: Duration = .seconds(5), - operation: @escaping @Sendable () async throws -> Void - ) async throws { - while !Task.isCancelled { - do { - try Task.checkCancellation() - try await operation() - SDLLogger.log("[SDLContext] worker \(name) ended, will restart", category: .context) - } catch is CancellationError { - SDLLogger.log("[SDLContext] worker \(name) cancelled", category: .context) - throw CancellationError() - } catch { - SDLLogger.log("[SDLContext] worker \(name) crashed: \(error.localizedDescription), will restart", category: .context) - } - - try await Task.sleep(for: retryDelay) - } - } private static func runPeriodic( name: String, diff --git a/Tun/DNS/DNSCloudService.swift b/Tun/DNS/DNSCloudService.swift index e5edf92..e25fee5 100644 --- a/Tun/DNS/DNSCloudService.swift +++ b/Tun/DNS/DNSCloudService.swift @@ -5,9 +5,14 @@ actor DNSCloudService { private var onEvent: DNSEventHandler = { _ in } private var currentClient: DNSCloudClient? + private var isRunning = false + private var isStopping = false + private var needsImmediateRestart = false + private let retryDelay: Duration - init(serverIP: String) { + init(serverIP: String, retryDelay: Duration = .seconds(5)) { self.serverIP = serverIP + self.retryDelay = retryDelay } func updateEventHandler(_ onEvent: @escaping DNSEventHandler) { @@ -15,54 +20,69 @@ actor DNSCloudService { } func run() async throws { - let client = DNSCloudClient(serverIP: self.serverIP, port: 15353) - self.currentClient = client - - defer { - self.clearCurrent(client) + guard !self.isRunning else { + return } - do { - let onEvent = self.onEvent + self.isRunning = true + self.isStopping = false - try await withThrowingTaskGroup(of: Void.self) { group in - defer { - group.cancelAll() + defer { + self.isRunning = false + self.currentClient = nil + } + + while !Task.isCancelled, !self.isStopping { + let client = DNSCloudClient(serverIP: self.serverIP, port: 15353) + self.currentClient = client + + do { + try await self.run(client: client) + self.clearCurrent(client) + await client.stop() + + guard !self.isStopping else { + break } - group.addTask { - try await client.run() + SDLLogger.log("[DNSCloudService] dnsCloudClient ended, will restart", category: .dns) + } catch is CancellationError { + self.clearCurrent(client) + await client.stop() + throw CancellationError() + } catch { + self.clearCurrent(client) + await client.stop() + + guard !self.isStopping else { + break } - group.addTask { - for try await packet in client.packetFlow { - try Task.checkCancellation() - await onEvent(.packet(packet)) - } - } - - _ = try await group.next() + SDLLogger.log("[DNSCloudService] dnsCloudClient failed: \(error.localizedDescription), will restart", category: .dns) } - await client.stop() - } catch is CancellationError { - await client.stop() - throw CancellationError() - } catch { - await client.stop() - throw error + if self.consumeImmediateRestartRequest() { + SDLLogger.log("[DNSCloudService] dnsCloudClient invalidated after wakeup, will restart immediately", category: .dns) + continue + } + + try await Task.sleep(for: self.retryDelay) } } func stop() async { - let client = self.currentClient - self.currentClient = nil - - await client?.stop() + self.isStopping = true + self.needsImmediateRestart = false + await self.invalidateCurrentClient() } func recoverAfterWake() async { - await self.stop() + guard !self.isStopping else { + return + } + + self.needsImmediateRestart = self.currentClient != nil + await self.invalidateCurrentClient() } func forward(ipPacketData: Data) async { @@ -74,4 +94,40 @@ actor DNSCloudService { self.currentClient = nil } } + + private func invalidateCurrentClient() async { + let client = self.currentClient + self.currentClient = nil + + await client?.stop() + } + + private func consumeImmediateRestartRequest() -> Bool { + let needsImmediateRestart = self.needsImmediateRestart + self.needsImmediateRestart = false + return needsImmediateRestart + } + + private func run(client: DNSCloudClient) async throws { + let onEvent = self.onEvent + + try await withThrowingTaskGroup(of: Void.self) { group in + defer { + group.cancelAll() + } + + group.addTask { + try await client.run() + } + + group.addTask { + for try await packet in client.packetFlow { + try Task.checkCancellation() + await onEvent(.packet(packet)) + } + } + + _ = try await group.next() + } + } } diff --git a/Tun/DNS/DNSLocalService.swift b/Tun/DNS/DNSLocalService.swift index 2734e96..cec079f 100644 --- a/Tun/DNS/DNSLocalService.swift +++ b/Tun/DNS/DNSLocalService.swift @@ -5,9 +5,14 @@ actor DNSLocalService { private var onEvent: DNSEventHandler = { _ in } private var currentClient: DNSLocalClient? + private var isRunning = false + private var isStopping = false + private var needsImmediateRestart = false + private let retryDelay: Duration - init(publicDnsServers: [String]) { + init(publicDnsServers: [String], retryDelay: Duration = .seconds(5)) { self.publicDnsServers = publicDnsServers + self.retryDelay = retryDelay } func updateEventHandler(_ onEvent: @escaping DNSEventHandler) { @@ -15,36 +20,72 @@ actor DNSLocalService { } func run() async throws { - let dnsServer = self.publicDnsServers.randomElement() ?? "223.5.5.5" - let client = DNSLocalClient(host: dnsServer) - self.currentClient = client + guard !self.isRunning else { + return + } - SDLLogger.log("[DNSLocalService] dnsLocalClient started", category: .dns) + self.isRunning = true + self.isStopping = false - do { - try await self.run(client: client) - self.clearCurrent(client) - await client.stop() - } catch is CancellationError { - self.clearCurrent(client) - await client.stop() - throw CancellationError() - } catch { - self.clearCurrent(client) - await client.stop() - throw error + defer { + self.isRunning = false + self.currentClient = nil + } + + while !Task.isCancelled, !self.isStopping { + let dnsServer = self.publicDnsServers.randomElement() ?? "223.5.5.5" + let client = DNSLocalClient(host: dnsServer) + self.currentClient = client + + SDLLogger.log("[DNSLocalService] dnsLocalClient started", category: .dns) + + do { + try await self.run(client: client) + self.clearCurrent(client) + await client.stop() + + guard !self.isStopping else { + break + } + + SDLLogger.log("[DNSLocalService] dnsLocalClient ended, will restart", category: .dns) + } catch is CancellationError { + self.clearCurrent(client) + await client.stop() + throw CancellationError() + } catch { + self.clearCurrent(client) + await client.stop() + + guard !self.isStopping else { + break + } + + SDLLogger.log("[DNSLocalService] dnsLocalClient failed: \(error.localizedDescription), will restart", category: .dns) + } + + if self.consumeImmediateRestartRequest() { + SDLLogger.log("[DNSLocalService] dnsLocalClient invalidated after wakeup, will restart immediately", category: .dns) + continue + } + + try await Task.sleep(for: self.retryDelay) } } func stop() async { - let client = self.currentClient - self.currentClient = nil - - await client?.stop() + self.isStopping = true + self.needsImmediateRestart = false + await self.invalidateCurrentClient() } func recoverAfterWake() async { - await self.stop() + guard !self.isStopping else { + return + } + + self.needsImmediateRestart = self.currentClient != nil + await self.invalidateCurrentClient() } func query(tracker: DNSLocalClient.DNSTracker, dnsPayload: Data) async { @@ -79,4 +120,17 @@ actor DNSLocalService { self.currentClient = nil } } + + private func invalidateCurrentClient() async { + let client = self.currentClient + self.currentClient = nil + + await client?.stop() + } + + private func consumeImmediateRestartRequest() -> Bool { + let needsImmediateRestart = self.needsImmediateRestart + self.needsImmediateRestart = false + return needsImmediateRestart + } } diff --git a/Tun/Super/SDLSuperService.swift b/Tun/Super/SDLSuperService.swift index f2b5127..8e4ca76 100644 --- a/Tun/Super/SDLSuperService.swift +++ b/Tun/Super/SDLSuperService.swift @@ -53,6 +53,11 @@ actor SDLSuperService { try await session.run() self.clearCurrent(session, generation: generation) await session.stop() + + guard !self.isStopping else { + break + } + SDLLogger.log("[SDLSuperService] session ended, will restart", category: .super) } catch is CancellationError { self.clearCurrent(session, generation: generation) @@ -80,6 +85,7 @@ actor SDLSuperService { func stop() async { self.isStopping = true + self.needsImmediateRestart = false await self.invalidateCurrentSession() } @@ -88,7 +94,7 @@ actor SDLSuperService { return } - self.needsImmediateRestart = true + self.needsImmediateRestart = self.currentSession != nil await self.invalidateCurrentSession() } diff --git a/Tun/UDPHole/SDLUDPHoleService.swift b/Tun/UDPHole/SDLUDPHoleService.swift index 39f0ad4..3df332f 100644 --- a/Tun/UDPHole/SDLUDPHoleService.swift +++ b/Tun/UDPHole/SDLUDPHoleService.swift @@ -18,9 +18,14 @@ actor SDLUDPHoleService { private var onData: DataHandler = { _ in } private var currentSession: SDLUDPHoleSession? private var generation: UInt64 = 0 + private var isRunning = false + private var isStopping = false + private var needsImmediateRestart = false + private let retryDelay: Duration - init(proberActor: SDLNATProberActor) { + init(proberActor: SDLNATProberActor, retryDelay: Duration = .seconds(5)) { self.proberActor = proberActor + self.retryDelay = retryDelay } func updateHandlers(onEvent: @escaping EventHandler, onData: @escaping DataHandler) { @@ -29,44 +34,77 @@ actor SDLUDPHoleService { } func run() async throws { - let generation = self.nextGeneration() - let session = SDLUDPHoleSession( - proberActor: self.proberActor, - onEvent: { [weak self] event in - await self?.handleEvent(event, generation: generation) - }, - onData: self.onData - ) + guard !self.isRunning else { + return + } - self.currentSession = session + self.isRunning = true + self.isStopping = false - do { - try await session.run() - self.clearCurrent(session, generation: generation) - await session.stop() - } catch is CancellationError { - self.clearCurrent(session, generation: generation) - await session.stop() - throw CancellationError() - } catch { - self.clearCurrent(session, generation: generation) - await session.stop() - throw error + defer { + self.isRunning = false + self.currentSession = nil + } + + while !Task.isCancelled, !self.isStopping { + let generation = self.nextGeneration() + let session = SDLUDPHoleSession( + proberActor: self.proberActor, + onEvent: { [weak self] event in + await self?.handleEvent(event, generation: generation) + }, + onData: self.onData + ) + + self.currentSession = session + + do { + try await session.run() + self.clearCurrent(session, generation: generation) + await session.stop() + + guard !self.isStopping else { + break + } + + SDLLogger.log("[SDLUDPHoleService] session ended, will restart", category: .udpHole) + } catch is CancellationError { + self.clearCurrent(session, generation: generation) + await session.stop() + throw CancellationError() + } catch { + self.clearCurrent(session, generation: generation) + await session.stop() + + guard !self.isStopping else { + break + } + + SDLLogger.log("[SDLUDPHoleService] session failed: \(error.localizedDescription), will restart", category: .udpHole) + } + + if self.consumeImmediateRestartRequest() { + SDLLogger.log("[SDLUDPHoleService] session invalidated after wakeup, will restart immediately", category: .udpHole) + continue + } + + try await Task.sleep(for: self.retryDelay) } } func stop() async { - self.generation &+= 1 - - let session = self.currentSession - self.currentSession = nil - - await session?.stop() - await self.proberActor.cancelAll() + self.isStopping = true + self.needsImmediateRestart = false + await self.invalidateCurrentSession() } func recoverAfterWake() async { - await self.stop() + guard !self.isStopping else { + return + } + + self.needsImmediateRestart = self.currentSession != nil + await self.invalidateCurrentSession() } func send(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async { @@ -78,6 +116,22 @@ actor SDLUDPHoleService { return self.generation } + private func invalidateCurrentSession() async { + self.generation &+= 1 + + let session = self.currentSession + self.currentSession = nil + + await session?.stop() + await self.proberActor.cancelAll() + } + + private func consumeImmediateRestartRequest() -> Bool { + let needsImmediateRestart = self.needsImmediateRestart + self.needsImmediateRestart = false + return needsImmediateRestart + } + private func clearCurrent(_ session: SDLUDPHoleSession, generation: UInt64) { guard self.generation == generation else { return diff --git a/Tun/UDPHoleV6/SDLUDPHoleV6Service.swift b/Tun/UDPHoleV6/SDLUDPHoleV6Service.swift index fbae1f2..3dd7e52 100644 --- a/Tun/UDPHoleV6/SDLUDPHoleV6Service.swift +++ b/Tun/UDPHoleV6/SDLUDPHoleV6Service.swift @@ -9,6 +9,14 @@ actor SDLUDPHoleV6Service { private var onData: DataHandler = { _ in } private var currentSession: SDLUDPHoleV6Session? private var generation: UInt64 = 0 + private var isRunning = false + private var isStopping = false + private var needsImmediateRestart = false + private let retryDelay: Duration + + init(retryDelay: Duration = .seconds(5)) { + self.retryDelay = retryDelay + } func updateHandlers(onEvent: @escaping EventHandler, onData: @escaping DataHandler) { self.onEvent = onEvent @@ -16,42 +24,76 @@ actor SDLUDPHoleV6Service { } func run() async throws { - let generation = self.nextGeneration() - let session = SDLUDPHoleV6Session( - onEvent: { [weak self] event in - await self?.handleEvent(event, generation: generation) - }, - onData: self.onData - ) + guard !self.isRunning else { + return + } - self.currentSession = session + self.isRunning = true + self.isStopping = false - do { - try await session.run() - self.clearCurrent(session, generation: generation) - await session.stop() - } catch is CancellationError { - self.clearCurrent(session, generation: generation) - await session.stop() - throw CancellationError() - } catch { - self.clearCurrent(session, generation: generation) - await session.stop() - throw error + defer { + self.isRunning = false + self.currentSession = nil + } + + while !Task.isCancelled, !self.isStopping { + let generation = self.nextGeneration() + let session = SDLUDPHoleV6Session( + onEvent: { [weak self] event in + await self?.handleEvent(event, generation: generation) + }, + onData: self.onData + ) + + self.currentSession = session + + do { + try await session.run() + self.clearCurrent(session, generation: generation) + await session.stop() + + guard !self.isStopping else { + break + } + + SDLLogger.log("[SDLUDPHoleV6Service] session ended, will restart", category: .udpHole) + } catch is CancellationError { + self.clearCurrent(session, generation: generation) + await session.stop() + throw CancellationError() + } catch { + self.clearCurrent(session, generation: generation) + await session.stop() + + guard !self.isStopping else { + break + } + + SDLLogger.log("[SDLUDPHoleV6Service] session failed: \(error.localizedDescription), will restart", category: .udpHole) + } + + if self.consumeImmediateRestartRequest() { + SDLLogger.log("[SDLUDPHoleV6Service] session invalidated after wakeup, will restart immediately", category: .udpHole) + continue + } + + try await Task.sleep(for: self.retryDelay) } } func stop() async { - self.generation &+= 1 - - let session = self.currentSession - self.currentSession = nil - - await session?.stop() + self.isStopping = true + self.needsImmediateRestart = false + await self.invalidateCurrentSession() } func recoverAfterWake() async { - await self.stop() + guard !self.isStopping else { + return + } + + self.needsImmediateRestart = self.currentSession != nil + await self.invalidateCurrentSession() } func send(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async { @@ -63,6 +105,21 @@ actor SDLUDPHoleV6Service { return self.generation } + private func invalidateCurrentSession() async { + self.generation &+= 1 + + let session = self.currentSession + self.currentSession = nil + + await session?.stop() + } + + private func consumeImmediateRestartRequest() -> Bool { + let needsImmediateRestart = self.needsImmediateRestart + self.needsImmediateRestart = false + return needsImmediateRestart + } + private func clearCurrent(_ session: SDLUDPHoleV6Session, generation: UInt64) { guard self.generation == generation else { return