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

解决定时器的问题

Browse Source
This commit is contained in:
anlicheng 2026-02-03 16:07:30 +08:00
parent d964eb6e27
commit 55ea1cd09d
6 changed files with 205 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
Tun/PacketTunnelProvider.swift
View File

@ -13,7 +13,7 @@ enum TunnelError: Error {
}
class PacketTunnelProvider: NEPacketTunnelProvider {
var contextSupervisor: SDLContextSupervisor?
var contextActor: SDLContextActor?
private var rootTask: Task<Void, Error>?
override func startTunnel(options: [String: NSObject]?, completionHandler: @escaping (Error?) -> Void) {
@ -24,7 +24,7 @@ class PacketTunnelProvider: NEPacketTunnelProvider {
}
// 如果当前在运行状态，不允许重复请求
guard self.contextSupervisor == nil else {
guard self.contextActor == nil else {
completionHandler(TunnelError.invalidContext)
return
}
@ -35,22 +35,23 @@ class PacketTunnelProvider: NEPacketTunnelProvider {
let logger = SDLLogger(level: .debug)
self.rootTask = Task {
self.contextSupervisor = SDLContextSupervisor()
await self.contextSupervisor?.start(provider: self, config: config, rsaCipher: rsaCipher, aesCipher: aesChiper, logger: logger)
self.contextActor = SDLContextActor(provider: self, config: config, rsaCipher: rsaCipher, aesCipher: aesChiper, logger: logger)
await self.contextActor?.start()
completionHandler(nil)
}
}
override func stopTunnel(with reason: NEProviderStopReason, completionHandler: @escaping () -> Void) {
// Add code here to start the process of stopping the tunnel.
self.rootTask?.cancel()
Task {
await self.contextSupervisor?.stop()
}
self.contextSupervisor = nil
self.rootTask = nil
await self.contextActor?.stop()
self.contextActor = nil
completionHandler()
self.rootTask?.cancel()
self.rootTask = nil
completionHandler()
}
}
override func handleAppMessage(_ messageData: Data, completionHandler: ((Data?) -> Void)?) {

44
Tun/Punchnet/Actors/SDLContextSupervisor.swift
View File

@ -1,44 +0,0 @@
//
// SDLContextSupervisor.swift
// Tun
//
// Created by 安礼成 on 2026/2/2.
//
import Foundation
import NetworkExtension
actor SDLContextSupervisor {
private var context: SDLContextActor?
private var tasks: [Task<Void, Never>] = []
public func start(provider: NEPacketTunnelProvider, config: SDLConfiguration, rsaCipher: RSACipher, aesCipher: AESCipher, logger: SDLLogger) async {
let context = SDLContextActor(provider: provider, config: config, rsaCipher: rsaCipher, aesCipher: aesCipher, logger: logger)
self.context = context
tasks.append(spawnLoop { try await context.startNoticeClient()})
tasks.append(spawnLoop { try await context.startUDPHole()})
tasks.append(spawnLoop { try await context.startDnsClient()})
tasks.append(spawnLoop { try await context.startMonitor()})
}
func stop() {
tasks.forEach {$0.cancel()}
tasks.removeAll()
}
private func spawnLoop(_ body: @escaping () async throws -> Void) -> Task<Void, Never> {
return Task.detached {
while !Task.isCancelled {
do {
try await body()
} catch is CancellationError {
break
} catch {
try? await Task.sleep(nanoseconds: 2_000_000_000)
}
}
}
}
}

29
Tun/Punchnet/SDLAsyncTimerStream.swift Normal file
View File

@ -0,0 +1,29 @@
//
// SDLAsyncTimerStream.swift
// Tun
//
// Created by 安礼成 on 2026/2/3.
//
import Foundation
class SDLAsyncTimerStream {
let timer: DispatchSourceTimer
init() {
self.timer = DispatchSource.makeTimerSource(queue: .global())
}
func start(_ cont: AsyncStream<Void>.Continuation) {
timer.schedule(deadline: .now(), repeating: .seconds(5))
timer.setEventHandler {
cont.yield()
}
timer.resume()
}
deinit {
self.timer.cancel()
}
}

251
Tun/Punchnet/SDLContextActor.swift
View File

@ -29,11 +29,14 @@ actor SDLContextActor {
nonisolated let rsaCipher: RSACipher
// 依赖的变量
var udpHole: SDLUDPHole?
private var udpHole: SDLUDPHole?
private var udpHoleWorkers: [Task<Void, Never>]?
nonisolated let providerAdapter: SDLTunnelProviderAdapter
var puncherActor: SDLPuncherActor?
// dns的client对象
var dnsClient: SDLDNSClient?
private var dnsClient: SDLDNSClient?
private var dnsWorker: Task<Void, Never>?
// 网络探测对象
var proberActor: SDLNATProberActor?
@ -46,6 +49,7 @@ actor SDLContextActor {
// 网络状态变化的健康
private var monitor: SDLNetworkMonitor?
private var monitorWorker: Task<Void, Never>?
// 内部socket通讯
private var noticeClient: SDLNoticeClient?
@ -55,6 +59,9 @@ actor SDLContextActor {
nonisolated private let logger: SDLLogger
// 处理内部的需要长时间运行的任务
private var loopChildWorkers: [Task<Void, Never>] = []
public init(provider: NEPacketTunnelProvider, config: SDLConfiguration, rsaCipher: RSACipher, aesCipher: AESCipher, logger: SDLLogger) {
self.logger = logger
self.config = config
@ -66,126 +73,173 @@ actor SDLContextActor {
self.providerAdapter = SDLTunnelProviderAdapter(provider: provider, logger: logger)
}
public func startNoticeClient() async throws {
// 启动noticeClient
self.noticeClient = try SDLNoticeClient(noticePort: self.config.noticePort, logger: self.logger)
self.logger.log("[SDLContext] noticeClient started")
try await self.noticeClient?.waitClose()
public func start() {
self.startMonitor()
self.loopChildWorkers.append(spawnLoop {
let noticeClient = try self.startNoticeClient()
try await noticeClient.waitClose()
self.logger.log("[SDLContext] noticeClient closed!!!!")
})
self.loopChildWorkers.append(spawnLoop {
let dnsClient = try await self.startDnsClient()
try await dnsClient.waitClose()
self.logger.log("[SDLContext] dns closed!!!!")
})
self.loopChildWorkers.append(spawnLoop {
let udpHole = try await self.startUDPHole()
try await udpHole.waitClose()
self.logger.log("[SDLContext] udp closed!!!!")
})
}
public func startMonitor() async throws {
private func startNoticeClient() throws -> SDLNoticeClient {
// 启动noticeClient
let noticeClient = try SDLNoticeClient(noticePort: self.config.noticePort, logger: self.logger)
self.logger.log("[SDLContext] noticeClient started")
self.noticeClient = noticeClient
return noticeClient
}
private func startMonitor() {
self.monitorWorker?.cancel()
self.monitorWorker = nil
// 启动monitor
let monitor = SDLNetworkMonitor()
monitor.start()
self.logger.log("[SDLContext] monitor started")
self.monitor = monitor
for await event in monitor.eventStream {
switch event {
case .changed:
// 需要重新探测网络的nat类型
//self.natType = await self.getNatType()
self.logger.log("didNetworkPathChanged, nat type is: \(self.natType)", level: .info)
case .unreachable:
self.logger.log("didNetworkPathUnreachable", level: .warning)
self.monitorWorker = Task {
for await event in monitor.eventStream {
switch event {
case .changed:
// 需要重新探测网络的nat类型
//self.natType = await self.getNatType()
self.logger.log("didNetworkPathChanged, nat type is: \(self.natType)", level: .info)
case .unreachable:
self.logger.log("didNetworkPathUnreachable", level: .warning)
}
}
}
}
public func startDnsClient() async throws {
private func startDnsClient() async throws -> SDLDNSClient {
self.dnsWorker?.cancel()
self.dnsWorker = nil
// 启动dns服务
let dnsSocketAddress = try SocketAddress.makeAddressResolvingHost(self.config.remoteDnsServer, port: 15353)
let dnsClient = try await SDLDNSClient(dnsServerAddress: dnsSocketAddress, logger: self.logger)
let channel = try dnsClient.start()
try dnsClient.start()
self.logger.log("[SDLContext] dnsClient started")
self.dnsClient = dnsClient
try await withThrowingTaskGroup(of: Void.self) {group in
group.addTask {
// 处理事件流
for await packet in dnsClient.packetFlow {
try Task.checkCancellation()
let nePacket = NEPacket(data: packet, protocolFamily: 2)
self.providerAdapter.writePackets(packets: [nePacket])
self.dnsWorker = Task {
// 处理事件流
for await packet in dnsClient.packetFlow {
if Task.isCancelled {
break
}
let nePacket = NEPacket(data: packet, protocolFamily: 2)
self.providerAdapter.writePackets(packets: [nePacket])
}
group.addTask {
try await channel.closeFuture.get()
}
try await group.next()
self.logger.log("[SDLContext] taskGroup cancel")
group.cancelAll()
}
return dnsClient
}
public func startUDPHole() async throws {
private func startUDPHole() async throws -> SDLUDPHole {
self.udpHoleWorkers?.forEach {$0.cancel()}
self.udpHoleWorkers = nil
// 启动udp服务器
let udpHole = try SDLUDPHole(logger: self.logger)
let channel = try udpHole.start()
try udpHole.start()
self.logger.log("[SDLContext] udpHole started")
self.udpHole = udpHole
await udpHole.channelIsActived()
await self.handleUDPHoleReady()
try await withThrowingTaskGroup(of: Void.self) { group in
group.addTask {
try await channel.closeFuture.get()
}
// 处理心跳逻辑
let pingTask = Task.detached {
let (stream, cont) = AsyncStream.makeStream(of: Void.self)
let timerStream = SDLAsyncTimerStream()
timerStream.start(cont)
// 处理UDP的事件流
group.addTask {
while true {
try Task.checkCancellation()
try await Task.sleep(for: .seconds(5))
await self.sendStunRequest()
for await _ in stream {
if Task.isCancelled {
break
}
self.logger.log("[SDLContext] will do stunRequest22")
await self.sendStunRequest()
self.logger.log("[SDLContext] will do stunRequest44")
}
// 处理数据流
group.addTask {
for try await data in udpHole.dataStream {
try Task.checkCancellation()
try await self.handleData(data: data)
}
}
// 处理signal信号流
group.addTask {
for try await(remoteAddress, signal) in udpHole.signalStream {
try Task.checkCancellation()
switch signal {
case .registerSuperAck(let registerSuperAck):
await self.handleRegisterSuperAck(registerSuperAck: registerSuperAck)
case .registerSuperNak(let registerSuperNak):
await self.handleRegisterSuperNak(nakPacket: registerSuperNak)
case .peerInfo(let peerInfo):
await self.puncherActor?.handlePeerInfo(peerInfo: peerInfo)
case .event(let event):
try await self.handleEvent(event: event)
case .stunProbeReply(let probeReply):
await self.proberActor?.handleProbeReply(reply: probeReply)
case .register(let register):
try await self.handleRegister(remoteAddress: remoteAddress, register: register)
case .registerAck(let registerAck):
await self.handleRegisterAck(remoteAddress: remoteAddress, registerAck: registerAck)
}
}
}
try await group.next()
group.cancelAll()
self.logger.log("[SDLContext] taskGroup cancel")
self.logger.log("[SDLContext] will do stunRequest55")
}
// 处理数据流
let dataTask = Task {
for await data in udpHole.dataStream {
if Task.isCancelled {
break
}
try? self.handleData(data: data)
}
}
// 处理控制信号
let signalTask = Task {
for await(remoteAddress, signal) in udpHole.signalStream {
if Task.isCancelled {
break
}
switch signal {
case .registerSuperAck(let registerSuperAck):
await self.handleRegisterSuperAck(registerSuperAck: registerSuperAck)
case .registerSuperNak(let registerSuperNak):
self.handleRegisterSuperNak(nakPacket: registerSuperNak)
case .peerInfo(let peerInfo):
await self.puncherActor?.handlePeerInfo(peerInfo: peerInfo)
case .event(let event):
try? self.handleEvent(event: event)
case .stunProbeReply(let probeReply):
await self.proberActor?.handleProbeReply(reply: probeReply)
case .register(let register):
try? self.handleRegister(remoteAddress: remoteAddress, register: register)
case .registerAck(let registerAck):
self.handleRegisterAck(remoteAddress: remoteAddress, registerAck: registerAck)
}
}
}
self.udpHoleWorkers = [pingTask, dataTask, signalTask]
return udpHole
}
// 处理context的停止问题
public func stop() async {
self.udpHole = nil
self.noticeClient = nil
self.loopChildWorkers.forEach { $0.cancel() }
self.loopChildWorkers.removeAll()
self.udpHoleWorkers?.forEach { $0.cancel() }
self.udpHoleWorkers = nil
self.dnsWorker?.cancel()
self.dnsWorker = nil
self.monitorWorker?.cancel()
self.monitorWorker = nil
self.readTask?.cancel()
self.readTask = nil
}
private func setNatType(natType: SDLNATProberActor.NatType) {
@ -227,7 +281,7 @@ actor SDLContextActor {
}
}
private func sendStunRequest() async {
private func sendStunRequest() {
var stunRequest = SDLStunRequest()
stunRequest.clientID = self.config.clientId
stunRequest.networkID = self.config.networkAddress.networkId
@ -418,21 +472,8 @@ actor SDLContextActor {
let packets = await self.providerAdapter.readPackets()
let ipPackets = packets.compactMap { IPPacket($0) }
await self.batchProcessPackets(batchSize: 20, packets: ipPackets)
}
}
}
// 批量分发ip数据包
private func batchProcessPackets(batchSize: Int, packets: [IPPacket]) async {
for startIndex in stride(from: 0, to: packets.count, by: batchSize) {
let endIndex = Swift.min(startIndex + batchSize, packets.count)
let chunkPackets = packets[startIndex..<endIndex]
await withDiscardingTaskGroup() { group in
for packet in chunkPackets {
group.addTask {
await self.dealPacket(packet: packet)
}
for ipPacket in ipPackets {
self.dealPacket(packet: ipPacket)
}
}
}
@ -508,6 +549,20 @@ actor SDLContextActor {
}
}
private func spawnLoop(_ body: @escaping () async throws -> Void) -> Task<Void, Never> {
return Task.detached {
while !Task.isCancelled {
do {
try await body()
} catch is CancellationError {
break
} catch {
try? await Task.sleep(nanoseconds: 2_000_000_000)
}
}
}
}
deinit {
self.udpHole = nil
self.dnsClient = nil

6
Tun/Punchnet/SDLDNSClient.swift
View File

@ -29,7 +29,7 @@ final class SDLDNSClient: ChannelInboundHandler {
(self.packetFlow, self.packetContinuation) = AsyncStream.makeStream(of: Data.self, bufferingPolicy: .unbounded)
}
func start() throws -> Channel {
func start() throws {
let bootstrap = DatagramBootstrap(group: group)
.channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
.channelInitializer { channel in
@ -39,8 +39,10 @@ final class SDLDNSClient: ChannelInboundHandler {
let channel = try bootstrap.bind(host: "0.0.0.0", port: 0).wait()
self.logger.log("[DNSClient] started", level: .debug)
self.channel = channel
}
return channel
func waitClose() async throws {
try await self.channel?.closeFuture.get()
}
// --MARK: ChannelInboundHandler delegate

8
Tun/Punchnet/SDLUDPHole.swift
View File

@ -49,7 +49,7 @@ final class SDLUDPHole: ChannelInboundHandler {
(self.dataStream, self.dataContinuation) = AsyncStream.makeStream(of: SDLData.self, bufferingPolicy: .unbounded)
}
func start() throws -> Channel {
func start() throws {
let bootstrap = DatagramBootstrap(group: group)
.channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
.channelInitializer { channel in
@ -59,8 +59,6 @@ final class SDLUDPHole: ChannelInboundHandler {
let channel = try bootstrap.bind(host: "0.0.0.0", port: 0).wait()
self.logger.log("[UDPHole] started", level: .debug)
self.channel = channel
return channel
}
func channelIsActived() async {
@ -73,6 +71,10 @@ final class SDLUDPHole: ChannelInboundHandler {
}
}
func waitClose() async throws {
try await self.channel?.closeFuture.get()
}
// --MARK: ChannelInboundHandler delegate
func channelActive(context: ChannelHandlerContext) {