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-05 16:40:16 +08:00
parent 3bec145f23
commit 385edf3c6c
4 changed files with 33 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Tun/Punchnet/Actors/ArpServer.swift
View File

@ -82,7 +82,7 @@ actor ArpServer {
var arpRequest = SDLArpRequest() var arpRequest = SDLArpRequest()
arpRequest.targetIp = targetIp arpRequest.targetIp = targetIp
quicClient.send(type: .arpRequest, data: try arpRequest.serializedData()) await quicClient.send(type: .arpRequest, data: try arpRequest.serializedData())
} }
func handleArpResponse(arpResponse: SDLArpResponse) { func handleArpResponse(arpResponse: SDLArpResponse) {

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

@ -166,12 +166,14 @@ actor SDLContextActor {
// 启动monitor // 启动monitor
let quicClient = SDLQUICClient(host: self.config.serverHost, port: 1443) let quicClient = SDLQUICClient(host: self.config.serverHost, port: 1443)
self.quicClient = quicClient self.quicClient = quicClient
quicClient.start() await quicClient.start()
defer { defer {
self.quicClient?.stop() Task {
self.quicClient = nil await self.quicClient?.stop()
SDLLogger.log("[SDLContext] quicClient: stop") self.quicClient = nil
SDLLogger.log("[SDLContext] quicClient: stop")
}
} }
// 这里必须等待quic的协商完成 // 这里必须等待quic的协商完成
@ -185,7 +187,7 @@ actor SDLContextActor {
} }
group.addTask { group.addTask {
for try await message in quicClient.messageStream { for try await message in await quicClient.messageStream {
try Task.checkCancellation() try Task.checkCancellation()
await self.handleQUICMessage(message: message) await self.handleQUICMessage(message: message)
} }
@ -195,7 +197,7 @@ actor SDLContextActor {
while true { while true {
try await Task.sleep(for: .seconds(5)) try await Task.sleep(for: .seconds(5))
try Task.checkCancellation() try Task.checkCancellation()
quicClient.send(type: .ping, data: Data()) await quicClient.send(type: .ping, data: Data())
} }
SDLLogger.log("[SDLQUICClient] udp pingTask cancel", for: .debug) SDLLogger.log("[SDLQUICClient] udp pingTask cancel", for: .debug)
} }
@ -203,7 +205,9 @@ actor SDLContextActor {
try await group.next() try await group.next()
} }
} onCancel: { } onCancel: {
quicClient.stop() Task {
await quicClient.stop()
}
} }
} }
@ -853,7 +857,7 @@ extension SDLContextActor {
if let registerSuperData = try? registerSuper.serializedData() { if let registerSuperData = try? registerSuper.serializedData() {
SDLLogger.log("[SDLContext] will send register super") SDLLogger.log("[SDLContext] will send register super")
self.quicClient?.send(type: .registerSuper, data: registerSuperData) await self.quicClient?.send(type: .registerSuper, data: registerSuperData)
} }
} }

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

@ -90,7 +90,7 @@ actor SDLPuncherActor {
phase: .waitingPeerInfo(deadline: now.addingTimeInterval(self.peerInfoTimeout)) phase: .waitingPeerInfo(deadline: now.addingTimeInterval(self.peerInfoTimeout))
) )
quicClient.send(type: .queryInfo, data: queryData) await quicClient.send(type: .queryInfo, data: queryData)
} }
func handlePeerInfo(using udpHole: SDLUDPHole?, udpHoleV6: SDLUDPHoleV6?, peerInfo: SDLPeerInfo) async { func handlePeerInfo(using udpHole: SDLUDPHole?, udpHoleV6: SDLUDPHoleV6?, peerInfo: SDLPeerInfo) async {

53
Tun/Punchnet/Actors/SDLQuicClient.swift
View File

@ -25,7 +25,7 @@ enum SDLQUICError: Error {
case dataStreamClosed case dataStreamClosed
} }
final class SDLQUICClient { actor SDLQUICClient {
enum State { enum State {
case idle case idle
case running case running
@ -45,7 +45,6 @@ final class SDLQUICClient {
private var connection: NWConnection? private var connection: NWConnection?
private let queue = DispatchQueue(label: "com.sdl.QUICClient.queue") // 专用队列保证线程安全 private let queue = DispatchQueue(label: "com.sdl.QUICClient.queue") // 专用队列保证线程安全
private let queueKey = DispatchSpecificKey<Void>()
private let host: String private let host: String
private let port: UInt16 private let port: UInt16
@ -56,8 +55,6 @@ final class SDLQUICClient {
self.frameParser = SDLQUICFrameParser(maxBufferSize: maxBufferSize) self.frameParser = SDLQUICFrameParser(maxBufferSize: maxBufferSize)
(self.messageStream, self.messageCont) = AsyncThrowingStream.makeStream(of: SDLQUICInboundMessage.self) (self.messageStream, self.messageCont) = AsyncThrowingStream.makeStream(of: SDLQUICInboundMessage.self)
self.queue.setSpecific(key: self.queueKey, value: ())
} }
func start() { func start() {
@ -86,16 +83,8 @@ final class SDLQUICClient {
connection.stateUpdateHandler = { [weak self] state in connection.stateUpdateHandler = { [weak self] state in
SDLLogger.log("[SDLQUICClient] new state: \(state)", for: .debug) SDLLogger.log("[SDLQUICClient] new state: \(state)", for: .debug)
switch state { Task {
case .ready: await self?.handleConnectionState(state: state)
self?.startReadTask()
self?.state = .running
case .failed(let error):
self?.finishMessageContinuationIfNeed(throwing: .connectionFailed(error))
case .cancelled:
self?.finishMessageContinuationIfNeed(throwing: .connectionCancelled)
default:
()
} }
} }
connection.start(queue: self.queue) connection.start(queue: self.queue)
@ -103,17 +92,21 @@ final class SDLQUICClient {
self.connection = connection self.connection = connection
} }
private func finishMessageContinuationIfNeed(throwing error: SDLQUICError?) { private func handleConnectionState(state: NWConnection.State) {
if DispatchQueue.getSpecific(key: queueKey) != nil { switch state {
self.finishMessageContinuationIfNeedOnQueue(throwing: error) case .ready:
} else { self.startReadTask()
queue.async { [weak self] in self.state = .running
self?.finishMessageContinuationIfNeedOnQueue(throwing: error) case .failed(let error):
} self.finishMessageContinuationIfNeed(throwing: .connectionFailed(error))
case .cancelled:
self.finishMessageContinuationIfNeed(throwing: .connectionCancelled)
default:
()
} }
} }
private func finishMessageContinuationIfNeedOnQueue(throwing error: SDLQUICError?) { private func finishMessageContinuationIfNeed(throwing error: SDLQUICError?) {
guard !self.isMessageContinuationFinished else { guard !self.isMessageContinuationFinished else {
return return
} }
@ -160,8 +153,10 @@ final class SDLQUICClient {
connection.send(content: packet, completion: .contentProcessed { [weak self] error in connection.send(content: packet, completion: .contentProcessed { [weak self] error in
if let error { if let error {
SDLLogger.log("[SDLQUICClient] send data get error: \(error)", for: .debug) Task {
self?.finishMessageContinuationIfNeed(throwing: .writeFailed(error)) SDLLogger.log("[SDLQUICClient] send data get error: \(error)", for: .debug)
await self?.finishMessageContinuationIfNeed(throwing: .writeFailed(error))
}
} }
}) })
} }
@ -188,16 +183,6 @@ final class SDLQUICClient {
} }
func stop() { func stop() {
if DispatchQueue.getSpecific(key: queueKey) != nil {
self.stopOnQueue()
} else {
queue.sync {
self.stopOnQueue()
}
}
}
private func stopOnQueue() {
guard self.state != .stopped else { guard self.state != .stopped else {
return return
} }