From f04196ef345bd81c8307d8eadd08b2d28d8f5078 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Thu, 21 May 2026 18:44:59 +0800 Subject: [PATCH] fix udp --- Tun/Context/SDLContextActor.swift | 33 ++++---- Tun/Outbound/PacketOutboundActor.swift | 4 + Tun/UDPHole/SDLUDPHole.swift | 10 ++- Tun/UDPHole/SDLUDPHoleService.swift | 110 ++++++++++++++++++------- Tun/UDPHole/SDLUDPHoleV6.swift | 6 +- 5 files changed, 114 insertions(+), 49 deletions(-) diff --git a/Tun/Context/SDLContextActor.swift b/Tun/Context/SDLContextActor.swift index 2ed6736..09245df 100644 --- a/Tun/Context/SDLContextActor.swift +++ b/Tun/Context/SDLContextActor.swift @@ -164,12 +164,18 @@ actor SDLContextActor { await self.packetOutboundActor.updateDNSService(dnsService) await dnsService.start() - await self.udpHoleServiceProxy.bindInbound(self.packetInboundActor) { [weak self] event in + let udpHoleEventHandler = await self.udpHoleServiceProxy.makeEventHandler { [weak self] event in await self?.handleUDPHoleControlEvent(event) } - - let udpHoleEventHandler = await self.udpHoleServiceProxy.makeEventHandler() - let udpHoleService = SDLUDPHoleService(proberActor: self.proberActor, onEvent: udpHoleEventHandler) + let packetInboundActor = self.packetInboundActor + let udpHoleService = SDLUDPHoleService( + proberActor: self.proberActor, + datagramSender: self.udpHoleServiceProxy.datagramSender, + onEvent: udpHoleEventHandler, + onData: { data in + await packetInboundActor.handleData(data) + } + ) await self.udpHoleServiceProxy.replace(udpHoleService) await udpHoleService.start(includeV6: false) @@ -469,19 +475,14 @@ extension SDLContextActor { } } - private func handleUDPHolePacket(remoteAddress: SocketAddress, message: SDLHoleMessage, source: SDLUDPHoleKind) async { + private func handleUDPHolePacket(remoteAddress: SocketAddress, message: SDLHoleControlMessage, source: SDLUDPHoleKind) async { switch message { - case .control(let message): - switch message { - case .stunReply(_), .stunProbeReply(_): - SDLLogger.log("[SDLContext] get a stun reply", for: .debug) - case .register(let register): - try? await self.handleRegister(remoteAddress: remoteAddress, register: register, source: source) - case .registerAck(let registerAck): - await self.handleRegisterAck(remoteAddress: remoteAddress, registerAck: registerAck, source: source) - } - case .data: - SDLLogger.log("[SDLContext] unexpected data packet in control path", for: .debug) + case .stunReply(_), .stunProbeReply(_): + SDLLogger.log("[SDLContext] get a stun reply", for: .debug) + case .register(let register): + try? await self.handleRegister(remoteAddress: remoteAddress, register: register, source: source) + case .registerAck(let registerAck): + await self.handleRegisterAck(remoteAddress: remoteAddress, registerAck: registerAck, source: source) } } diff --git a/Tun/Outbound/PacketOutboundActor.swift b/Tun/Outbound/PacketOutboundActor.swift index 72d012b..9742122 100644 --- a/Tun/Outbound/PacketOutboundActor.swift +++ b/Tun/Outbound/PacketOutboundActor.swift @@ -263,6 +263,10 @@ actor PacketOutboundActor { } private func sendPacket(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) async { + if type == .data, self.udpHoleServiceProxy.sendFast(type: type, data: data, remoteAddress: remoteAddress) { + return + } + await self.udpHoleServiceProxy.send(type: type, data: data, remoteAddress: remoteAddress) } } diff --git a/Tun/UDPHole/SDLUDPHole.swift b/Tun/UDPHole/SDLUDPHole.swift index 86c674e..85399cc 100644 --- a/Tun/UDPHole/SDLUDPHole.swift +++ b/Tun/UDPHole/SDLUDPHole.swift @@ -26,8 +26,8 @@ actor SDLUDPHole { private var state: State = .idle private let udpHoleHandler: SDLUDPHoleHandler - init() throws { - self.udpHoleHandler = try SDLUDPHoleHandler() + init(datagramSender: SDLUDPHoleDatagramSender) throws { + self.udpHoleHandler = try SDLUDPHoleHandler(datagramSender: datagramSender) } func start() async throws -> SocketAddress { @@ -69,6 +69,7 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler { private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) private var channel: Channel? + private let datagramSender: SDLUDPHoleDatagramSender private let locker = NSLock() @@ -77,7 +78,8 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler { private var isMessageContinuationFinished: Bool = false // 启动函数 - init() throws { + init(datagramSender: SDLUDPHoleDatagramSender) throws { + self.datagramSender = datagramSender let (stream, continuation) = AsyncThrowingStream.makeStream(of: (SocketAddress, SDLHoleMessage).self, bufferingPolicy: .bufferingNewest(2048)) self.messageStream = stream self.messageContinuation = continuation @@ -97,6 +99,7 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler { } self.channel = channel + self.datagramSender.bind(channel, for: .v4) return localAddress } @@ -154,6 +157,7 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler { self.finishMessageContinuationIfNeed(throwing: nil) let channel = self.channel self.channel = nil + self.datagramSender.bind(nil, for: .v4) try? channel?.close().wait() try? self.group.syncShutdownGracefully() diff --git a/Tun/UDPHole/SDLUDPHoleService.swift b/Tun/UDPHole/SDLUDPHoleService.swift index 9a61ec5..cd23fca 100644 --- a/Tun/UDPHole/SDLUDPHoleService.swift +++ b/Tun/UDPHole/SDLUDPHoleService.swift @@ -15,18 +15,69 @@ enum SDLUDPHoleKind: Equatable { } } +final class SDLUDPHoleDatagramSender: @unchecked Sendable { + private let lock = NSLock() + private var v4Channel: Channel? + private var v6Channel: Channel? + + func bind(_ channel: Channel?, for kind: SDLUDPHoleKind) { + self.lock.lock() + defer { + self.lock.unlock() + } + + switch kind { + case .v4: + self.v4Channel = channel + case .v6: + self.v6Channel = channel + } + } + + func send(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) -> Bool { + let channel: Channel? + self.lock.lock() + switch remoteAddress { + case .v4: + channel = self.v4Channel + case .v6: + channel = self.v6Channel + default: + channel = nil + } + self.lock.unlock() + + guard let channel else { + return false + } + + var buffer = channel.allocator.buffer(capacity: data.count + 1) + buffer.writeBytes([type.rawValue]) + buffer.writeBytes(data) + + let envelope = AddressedEnvelope(remoteAddress: remoteAddress, data: buffer) + channel.eventLoop.execute { + channel.writeAndFlush(envelope, promise: nil) + } + return true + } +} + actor SDLUDPHoleService { enum Event { case ready(SocketAddress) case natType(SDLNATProberActor.NatType) - case packet(SocketAddress, SDLHoleMessage, source: SDLUDPHoleKind) + case packet(SocketAddress, SDLHoleControlMessage, source: SDLUDPHoleKind) case closed(Error) } typealias EventHandler = @Sendable (Event) async -> Void + typealias DataHandler = @Sendable (SDLData) async -> Void private let proberActor: SDLNATProberActor private let onEvent: EventHandler + private let onData: DataHandler + private let datagramSender: SDLUDPHoleDatagramSender private var udpHole: SDLUDPHole? private var udpHoleMonitorTask: Task? @@ -36,9 +87,16 @@ actor SDLUDPHoleService { private var udpHoleV6: SDLUDPHoleV6? private var udpHoleV6MonitorTask: Task? - init(proberActor: SDLNATProberActor, onEvent: @escaping EventHandler) { + init( + proberActor: SDLNATProberActor, + datagramSender: SDLUDPHoleDatagramSender, + onEvent: @escaping EventHandler, + onData: @escaping DataHandler + ) { self.proberActor = proberActor + self.datagramSender = datagramSender self.onEvent = onEvent + self.onData = onData } func start(includeV6: Bool = false) { @@ -115,7 +173,7 @@ actor SDLUDPHoleService { } private func runV4() async throws { - let udpHole = try SDLUDPHole() + let udpHole = try SDLUDPHole(datagramSender: self.datagramSender) let localAddress = try await udpHole.start() self.udpHole = udpHole self.localAddress = localAddress @@ -171,10 +229,10 @@ actor SDLUDPHoleService { case .stunProbeReply(let probeReply): await self.proberActor.handleProbeReply(localAddress: self.localAddress, reply: probeReply) default: - await self.onEvent(.packet(remoteAddress, message, source: .v4)) + await self.onEvent(.packet(remoteAddress, control, source: .v4)) } - case .data: - await self.onEvent(.packet(remoteAddress, message, source: .v4)) + case .data(let data): + await self.onData(data) } } @@ -192,7 +250,7 @@ actor SDLUDPHoleService { } private func runV6() async throws { - let udpHoleV6 = try SDLUDPHoleV6() + let udpHoleV6 = try SDLUDPHoleV6(datagramSender: self.datagramSender) let localAddress = try udpHoleV6.start() self.udpHoleV6 = udpHoleV6 @@ -215,10 +273,16 @@ actor SDLUDPHoleService { } let onEvent = self.onEvent + let onData = self.onData group.addTask { for await (remoteAddress, message) in udpHoleV6.messageStream { try Task.checkCancellation() - await onEvent(.packet(remoteAddress, message, source: .v6)) + switch message { + case .control(let control): + await onEvent(.packet(remoteAddress, control, source: .v6)) + case .data(let data): + await onData(data) + } } } @@ -242,22 +306,16 @@ actor SDLUDPHoleService { actor SDLUDPHoleServiceProxy { typealias ControlEventHandler = @Sendable (SDLUDPHoleService.Event) async -> Void + nonisolated let datagramSender = SDLUDPHoleDatagramSender() private var udpHoleService: SDLUDPHoleService? - private var packetInboundActor: PacketInboundActor? - private var onControlEvent: ControlEventHandler? private var generation: UInt64 = 0 - func bindInbound(_ packetInboundActor: PacketInboundActor, onControlEvent: @escaping ControlEventHandler) { - self.packetInboundActor = packetInboundActor - self.onControlEvent = onControlEvent - } - - func makeEventHandler() -> SDLUDPHoleService.EventHandler { + 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) + await self?.handleEvent(event, generation: generation, onControlEvent: onControlEvent) } } @@ -283,21 +341,15 @@ actor SDLUDPHoleServiceProxy { await self.udpHoleService?.send(type: type, data: data, remoteAddress: remoteAddress) } - private func handleEvent(_ event: SDLUDPHoleService.Event, generation: UInt64) async { + nonisolated func sendFast(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) -> Bool { + self.datagramSender.send(type: type, data: data, remoteAddress: remoteAddress) + } + + private func handleEvent(_ event: SDLUDPHoleService.Event, generation: UInt64, onControlEvent: ControlEventHandler) async { guard generation == self.generation else { return } - switch event { - case .packet(_, let message, _): - switch message { - case .data(let data): - await self.packetInboundActor?.handleData(data) - case .control: - await self.onControlEvent?(event) - } - case .ready, .natType, .closed: - await self.onControlEvent?(event) - } + await onControlEvent(event) } } diff --git a/Tun/UDPHole/SDLUDPHoleV6.swift b/Tun/UDPHole/SDLUDPHoleV6.swift index a82d747..cff548e 100644 --- a/Tun/UDPHole/SDLUDPHoleV6.swift +++ b/Tun/UDPHole/SDLUDPHoleV6.swift @@ -25,6 +25,7 @@ final class SDLUDPHoleV6: ChannelInboundHandler { private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) private var channel: Channel? + private let datagramSender: SDLUDPHoleDatagramSender public let messageStream: AsyncStream<(SocketAddress, SDLHoleMessage)> private let messageContinuation: AsyncStream<(SocketAddress, SDLHoleMessage)>.Continuation @@ -34,7 +35,8 @@ final class SDLUDPHoleV6: ChannelInboundHandler { private let eventContinuation: AsyncStream.Continuation // 启动函数 - init() throws { + init(datagramSender: SDLUDPHoleDatagramSender) throws { + self.datagramSender = datagramSender let (stream, continuation) = AsyncStream.makeStream(of: (SocketAddress, SDLHoleMessage).self, bufferingPolicy: .bufferingNewest(2048)) self.messageStream = stream self.messageContinuation = continuation @@ -54,6 +56,7 @@ final class SDLUDPHoleV6: ChannelInboundHandler { // 绑定到IPv6通配地址，只处理IPv6流量 let channel = try bootstrap.bind(host: "::", port: 0).wait() self.channel = channel + self.datagramSender.bind(channel, for: .v6) return channel.localAddress } @@ -121,6 +124,7 @@ final class SDLUDPHoleV6: ChannelInboundHandler { let channel = self.channel self.channel = nil + self.datagramSender.bind(nil, for: .v6) try? channel?.close().wait() try? self.group.syncShutdownGracefully()