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-05-27 16:21:09 +08:00
parent c19b916365
commit 875a4c3669
10 changed files with 92 additions and 333 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
Tun/Context/SDLContextActor.swift
View File

@ -55,6 +55,8 @@ actor SDLContextActor {
// 本地ipv6地址信息探测
private var ipv6AssistClient: SDLIPV6AssistClient?
private let ipv6AssistEvents: AsyncStream<SDLV6Info?>
private let ipv6AssistContinuation: AsyncStream<SDLV6Info?>.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<SDLV6Info>.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)
}
// 注册

22
Tun/DNS/DNSService.swift
View File

@ -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))
}
}

45
Tun/Foundation/SDLThrottler.swift
View File

@ -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
}
}
}

54
Tun/NWPath/SDLPathDebounceActor.swift
View File

@ -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<Void, Never>?
private let delay: Duration
public let statusStream: AsyncStream<NWPath.Status>
private let statusCont: AsyncStream<NWPath.Status>.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
}
}

45
Tun/NWPath/SDLPathMonitor.swift
View File

@ -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<NWPath.Status> {
return self.debouncer.statusStream
}
func stop() {
Task {
await debouncer.stop()
}
monitor.cancel()
}
}

32
Tun/NWPath/SDLPathSnapshot.swift
View File

@ -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)
}
}

50
Tun/Network/SDLIPV6AssistClient.swift
View File

@ -26,7 +26,6 @@ actor SDLIPV6AssistClient {
private var state: State = .idle
private var connection: NWConnection?
private var receiveTask: Task<Void, Never>?
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

74
Tun/Network/SDLNetworkMonitor.swift
View File

@ -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<NWInterface.InterfaceType, Never>()
private var cancel: AnyCancellable?
private var isStopped = false
public let eventStream: AsyncStream<MonitorEvent>
private let eventContinuation: AsyncStream<MonitorEvent>.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()
}
}

9
Tun/Super/SDLSuperService.swift
View File

@ -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()

30
Tun/UDPHole/SDLUDPHoleService.swift
View File

@ -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()