anlicheng/punchnet-macos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tun/Punchnet/SDLUDPHoleV6.swift

Browse Source
This commit is contained in:
anlicheng 2026-04-15 17:14:46 +08:00
parent 2f0f1f6c7c
commit 5551d38b88
4 changed files with 118 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
Tun/Punchnet/Actors/SDLContextActor.swift
View File

@ -22,6 +22,11 @@ actor SDLContextActor {
case stopped
}
private enum UDPHoleKind: Equatable {
case v4
case v6
}
private var readyState: ReadyState = .idle
private var readyWaiters: [CheckedContinuation<Void, Error>] = []
@ -43,6 +48,9 @@ actor SDLContextActor {
private var udpHole: SDLUDPHole?
private var udpHoleWorkers: [Task<Void, Never>]?
private var udpHoleLocalAddress: SocketAddress?
private var udpHoleV6: SDLUDPHoleV6?
private var udpHoleV6Workers: [Task<Void, Never>]?
private var udpHoleV6LocalAddress: SocketAddress?
// dns的client对象
private var dnsClient: DNSCloudClient?
@ -137,6 +145,13 @@ actor SDLContextActor {
try await udpHole.waitClose()
SDLLogger.log("[SDLContext] udp closed!!!!")
}
await self.supervisor.addWorker(name: "udpHoleV6") {
let udpHoleV6 = try await self.startUDPHoleV6()
SDLLogger.log("[SDLContext] udp v6 running!!!!")
try await udpHoleV6.waitClose()
SDLLogger.log("[SDLContext] udp v6 closed!!!!")
}
}
public func waitForReady() async throws {
@ -194,7 +209,7 @@ actor SDLContextActor {
await self.handleRegisterSuperNak(nakPacket: registerSuperNak)
case .peerInfo(let peerInfo):
//SDLLogger.shared.log("[SDLContext] peer message: \(peerInfo)")
await self.puncherActor.handlePeerInfo(using: self.udpHole, peerInfo: peerInfo)
await self.puncherActor.handlePeerInfo(using: self.udpHole, udpHoleV6: self.udpHoleV6, peerInfo: peerInfo)
case .event(let event):
await self.handleEvent(event: event)
case .policyReponse(let policyResponse):
@ -316,7 +331,7 @@ actor SDLContextActor {
// 处理消息流
let messageStream = udpHole.messageStream
let messageTask = Task.detached {
await self.consumeUDPHoleMessages(stream: messageStream, localAddress: localAddress)
await self.consumeUDPHoleMessages(stream: messageStream, localAddress: localAddress, source: .v4)
}
self.udpHole = udpHole
@ -329,6 +344,28 @@ actor SDLContextActor {
return udpHole
}
private func startUDPHoleV6() async throws -> SDLUDPHoleV6 {
self.udpHoleV6Workers?.forEach {$0.cancel()}
self.udpHoleV6Workers = nil
// 启动udp服务器
let udpHoleV6 = try SDLUDPHoleV6()
let localAddress = try udpHoleV6.start()
SDLLogger.log("[SDLContext] udpHoleV6 started, on address: \(localAddress)")
// 处理消息流
let messageStream = udpHoleV6.messageStream
let messageTask = Task.detached {
await self.consumeUDPHoleMessages(stream: messageStream, localAddress: localAddress, source: .v6)
}
self.udpHoleV6 = udpHoleV6
self.udpHoleV6LocalAddress = localAddress
self.udpHoleV6Workers = [messageTask]
return udpHoleV6
}
// 处理context的停止问题
public func stop() async {
self.resumeReadyWaiters(.failure(CancellationError()))
@ -344,6 +381,12 @@ actor SDLContextActor {
self.udpHole = nil
self.udpHoleLocalAddress = nil
self.udpHoleV6Workers?.forEach { $0.cancel() }
self.udpHoleV6Workers = nil
self.udpHoleV6?.stop()
self.udpHoleV6 = nil
self.udpHoleV6LocalAddress = nil
self.quicWorker?.cancel()
self.quicWorker = nil
self.quicClient?.stop()
@ -502,16 +545,16 @@ actor SDLContextActor {
// 发送给super/stun节点的数据
private func sendSuperPacket(type: SDLPacketType, data: Data) {
self.udpHole?.send(type: type, data: data, remoteAddress: self.config.stunSocketAddress)
self.sendPacket(type: type, data: data, remoteAddress: self.config.stunSocketAddress)
}
// 发送给peer的数据
private func sendPeerPacket(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) {
self.udpHole?.send(type: type, data: data, remoteAddress: remoteAddress)
self.sendPacket(type: type, data: data, remoteAddress: remoteAddress)
}
private func makeCurrentV6Info() -> SDLV6Info? {
guard let port = self.udpHoleLocalAddress?.port else {
guard let port = self.udpHoleV6LocalAddress?.port else {
return nil
}
@ -535,6 +578,25 @@ actor SDLContextActor {
return v6Info
}
private func sendPacket(type: SDLPacketType, data: Data, remoteAddress: SocketAddress) {
switch remoteAddress {
case .v4:
guard let udpHole = self.udpHole else {
SDLLogger.log("[SDLContext] udpHole is nil for remoteAddress: \(remoteAddress)", for: .debug)
return
}
udpHole.send(type: type, data: data, remoteAddress: remoteAddress)
case .v6:
guard let udpHoleV6 = self.udpHoleV6 else {
SDLLogger.log("[SDLContext] udpHoleV6 is nil for remoteAddress: \(remoteAddress)", for: .debug)
return
}
udpHoleV6.send(type: type, data: data, remoteAddress: remoteAddress)
default:
SDLLogger.log("[SDLContext] unsupported socket family: \(remoteAddress)", for: .debug)
}
}
private func spawnLoop(_ body: @escaping () async throws -> Void) -> Task<Void, Never> {
return Task.detached {
while !Task.isCancelled {
@ -607,6 +669,8 @@ actor SDLContextActor {
deinit {
self.udpHole = nil
self.udpHoleLocalAddress = nil
self.udpHoleV6 = nil
self.udpHoleV6LocalAddress = nil
self.dnsClient = nil
}
}
@ -768,7 +832,7 @@ extension SDLContextActor {
// 处理从Hole收到的数据
extension SDLContextActor {
private func consumeUDPHoleMessages(stream: AsyncStream<(SocketAddress, SDLHoleMessage)>, localAddress: SocketAddress) async {
private func consumeUDPHoleMessages(stream: AsyncStream<(SocketAddress, SDLHoleMessage)>, localAddress: SocketAddress, source: UDPHoleKind) async {
for await (remoteAddress, message) in stream {
if Task.isCancelled {
break
@ -776,16 +840,19 @@ extension SDLContextActor {
switch message.inboundMessage {
case .control(let controlMessage):
await self.handleHoleControlMessage(controlMessage, localAddress: localAddress, remoteAddress: remoteAddress)
await self.handleHoleControlMessage(controlMessage, localAddress: localAddress, remoteAddress: remoteAddress, source: source)
case .data(let data):
try? await self.handleHoleData(data: data)
}
}
}
private func handleHoleControlMessage(_ message: SDLHoleControlMessage, localAddress: SocketAddress, remoteAddress: SocketAddress) async {
private func handleHoleControlMessage(_ message: SDLHoleControlMessage, localAddress: SocketAddress, remoteAddress: SocketAddress, source: UDPHoleKind) async {
switch message {
case .stunProbeReply(let probeReply):
guard source == .v4 else {
return
}
await self.proberActor.handleProbeReply(localAddress: localAddress, reply: probeReply)
case .register(let register):
try? self.handleRegister(remoteAddress: remoteAddress, register: register)

32
Tun/Punchnet/Actors/SDLPuncherActor.swift
View File

@ -93,7 +93,7 @@ actor SDLPuncherActor {
quicClient.send(type: .queryInfo, data: queryData)
}
func handlePeerInfo(using udpHole: SDLUDPHole?, peerInfo: SDLPeerInfo) async {
func handlePeerInfo(using udpHole: SDLUDPHole?, udpHoleV6: SDLUDPHoleV6?, peerInfo: SDLPeerInfo) async {
let now = Date()
self.cleanupExpiredEntries(now: now)
@ -108,8 +108,8 @@ actor SDLPuncherActor {
entry.markCoolingDown()
self.requestEntries[peerInfo.dstMac] = entry
guard let udpHole else {
SDLLogger.log("[SDLPuncherActor] udpHole is nil when peerInfo arrived", for: .debug)
guard udpHole != nil || udpHoleV6 != nil else {
SDLLogger.log("[SDLPuncherActor] udpHole and udpHoleV6 are nil when peerInfo arrived", for: .debug)
return
}
@ -127,7 +127,7 @@ actor SDLPuncherActor {
if peerInfo.hasV4Info {
if let remoteAddress = try? await peerInfo.v4Info.socketAddress() {
SDLLogger.log("[SDLContext] hole sock address: \(remoteAddress)", for: .debug)
udpHole.send(type: .register, data: registerData, remoteAddress: remoteAddress)
self.sendRegister(using: udpHole, udpHoleV6: udpHoleV6, registerData: registerData, remoteAddress: remoteAddress)
} else {
SDLLogger.log("[SDLPuncherActor] failed to resolve peerInfo.v4Info", for: .debug)
}
@ -136,7 +136,7 @@ actor SDLPuncherActor {
if peerInfo.hasV6Info {
if let remoteAddress = try? await peerInfo.v6Info.socketAddress() {
SDLLogger.log("[SDLContext] hole sock address v6: \(remoteAddress)", for: .debug)
udpHole.send(type: .register, data: registerData, remoteAddress: remoteAddress)
self.sendRegister(using: udpHole, udpHoleV6: udpHoleV6, registerData: registerData, remoteAddress: remoteAddress)
} else {
SDLLogger.log("[SDLPuncherActor] failed to resolve peerInfo.v6Info", for: .debug)
}
@ -156,6 +156,28 @@ actor SDLPuncherActor {
}
}
private func sendRegister(using udpHole: SDLUDPHole?,
udpHoleV6: SDLUDPHoleV6?,
registerData: Data,
remoteAddress: SocketAddress) {
switch remoteAddress {
case .v4:
guard let udpHole else {
SDLLogger.log("[SDLPuncherActor] udpHole is nil when v4 peerInfo arrived", for: .debug)
return
}
udpHole.send(type: .register, data: registerData, remoteAddress: remoteAddress)
case .v6:
guard let udpHoleV6 else {
SDLLogger.log("[SDLPuncherActor] udpHoleV6 is nil when v6 peerInfo arrived", for: .debug)
return
}
udpHoleV6.send(type: .register, data: registerData, remoteAddress: remoteAddress)
default:
SDLLogger.log("[SDLPuncherActor] unsupported peer address family: \(remoteAddress)", for: .debug)
}
}
deinit {
self.cleanupTask?.cancel()
}

8
Tun/Punchnet/SDLUDPHole.swift
View File

@ -44,8 +44,8 @@ final class SDLUDPHole: ChannelInboundHandler {
channel.pipeline.addHandler(self)
}
// 绑定到IPv6通配地址，依赖SwiftNIO创建dual-stack socket，同时接收IPv4/IPv6流量
let channel = try bootstrap.bind(host: "::", port: 0).wait()
// 绑定到IPv4通配地址，只处理IPv4流量
let channel = try bootstrap.bind(host: "0.0.0.0", port: 0).wait()
self.channel = channel
self.closeFuture = channel.closeFuture
self.state = .ready
@ -70,6 +70,7 @@ final class SDLUDPHole: ChannelInboundHandler {
}
func stop() {
SDLLogger.log("[SDLUDPHole] waitClose stop", for: .debug)
switch self.state {
case .stopping, .stopped:
return
@ -116,6 +117,7 @@ final class SDLUDPHole: ChannelInboundHandler {
self.finishMessageStream()
self.channel = nil
self.state = .stopped
SDLLogger.log("[SDLUDPHole] channelInactive", for: .debug)
}
func errorCaught(context: ChannelHandlerContext, error: any Error) {
@ -125,6 +127,7 @@ final class SDLUDPHole: ChannelInboundHandler {
self.state = .stopping
}
context.close(promise: nil)
SDLLogger.log("[SDLUDPHole] errorCaught", for: .debug)
}
// MARK: 处理写入逻辑
@ -198,6 +201,7 @@ final class SDLUDPHole: ChannelInboundHandler {
}
deinit {
SDLLogger.log("[SDLUDPHole] closeWait deinit", for: .debug)
self.stop()
try? self.group.syncShutdownGracefully()
}

2
tracelog.sh
View File

@ -1,3 +1,3 @@
#! /bin/sh
log stream --predicate 'subsystem == "com.jihe.punchnet"' --info --style compact
log stream --predicate 'subsystem == "com.jihe.punchnet.debug"' --info --style compact