fix udpHole
This commit is contained in:
parent
7f56806852
commit
0a7c8eca85
@ -166,6 +166,7 @@ actor SDLContextActor {
|
|||||||
|
|
||||||
defer {
|
defer {
|
||||||
self.quicClient = nil
|
self.quicClient = nil
|
||||||
|
SDLLogger.log("[SDLContext] startSuperClient defer")
|
||||||
}
|
}
|
||||||
|
|
||||||
// 这里必须等待quic的协商完成
|
// 这里必须等待quic的协商完成
|
||||||
@ -197,13 +198,15 @@ actor SDLContextActor {
|
|||||||
try await group.next()
|
try await group.next()
|
||||||
}
|
}
|
||||||
} onCancel: {
|
} onCancel: {
|
||||||
SDLLogger.log("[SDLQUICClient] quicClient taskGroup cancel", for: .debug)
|
SDLLogger.log("[SDLQUICClient] startSuperClient taskGroup cancel", for: .debug)
|
||||||
Task {
|
Task {
|
||||||
|
SDLLogger.log("[SDLContext] startSuperClient onCancel")
|
||||||
await quicClient.stop()
|
await quicClient.stop()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} catch let err {
|
} catch let err {
|
||||||
await quicClient.stop()
|
await quicClient.stop()
|
||||||
|
SDLLogger.log("[SDLContext] startSuperClient catch err: \(err)")
|
||||||
throw err
|
throw err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -192,6 +192,12 @@ actor SDLQUICClient {
|
|||||||
self.readTask?.cancel()
|
self.readTask?.cancel()
|
||||||
self.connection?.cancel()
|
self.connection?.cancel()
|
||||||
self.finishMessageContinuationIfNeed(throwing: nil)
|
self.finishMessageContinuationIfNeed(throwing: nil)
|
||||||
|
|
||||||
|
SDLLogger.log("[SDLQUICClient] stopped")
|
||||||
|
}
|
||||||
|
|
||||||
|
deinit {
|
||||||
|
SDLLogger.log("[SDLQUICClient] deinit")
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@ -88,6 +88,8 @@ final class DNSCloudClient {
|
|||||||
self.connection = nil
|
self.connection = nil
|
||||||
|
|
||||||
self.finishPacketContinuationIfNeed(throwing: nil)
|
self.finishPacketContinuationIfNeed(throwing: nil)
|
||||||
|
|
||||||
|
SDLLogger.log("[SDLCloudClient] stopped")
|
||||||
}
|
}
|
||||||
|
|
||||||
private func handleConnectionStateUpdate(_ state: NWConnection.State, for connection: NWConnection) {
|
private func handleConnectionStateUpdate(_ state: NWConnection.State, for connection: NWConnection) {
|
||||||
@ -156,4 +158,8 @@ final class DNSCloudClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
deinit {
|
||||||
|
SDLLogger.log("[DNSCloudClient] deinit", for: .debug)
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@ -132,6 +132,8 @@ actor DNSLocalClient {
|
|||||||
connection?.cancel()
|
connection?.cancel()
|
||||||
cleanupTask?.cancel()
|
cleanupTask?.cancel()
|
||||||
self.finishPacketContinuationIfNeed(throwing: nil)
|
self.finishPacketContinuationIfNeed(throwing: nil)
|
||||||
|
|
||||||
|
SDLLogger.log("[SDLLocalClient] stopped")
|
||||||
}
|
}
|
||||||
|
|
||||||
private func handleConnectionStateUpdate(_ state: NWConnection.State, for conn: NWConnection) {
|
private func handleConnectionStateUpdate(_ state: NWConnection.State, for conn: NWConnection) {
|
||||||
@ -292,6 +294,10 @@ actor DNSLocalClient {
|
|||||||
receiveNext()
|
receiveNext()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
deinit {
|
||||||
|
SDLLogger.log("[DNSLocalClient] deinit", for: .debug)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
extension DNSLocalClient {
|
extension DNSLocalClient {
|
||||||
|
|||||||
74
Tun/Punchnet/UDPHole/SDLUDPCounter.swift
Normal file
74
Tun/Punchnet/UDPHole/SDLUDPCounter.swift
Normal file
@ -0,0 +1,74 @@
|
|||||||
|
//
|
||||||
|
// SDLUDPCounter.swift
|
||||||
|
// punchnet
|
||||||
|
//
|
||||||
|
// Created by 安礼成 on 2026/5/6.
|
||||||
|
//
|
||||||
|
|
||||||
|
import Foundation
|
||||||
|
|
||||||
|
actor SDLUDPCounter {
|
||||||
|
|
||||||
|
enum Direction {
|
||||||
|
case inbound
|
||||||
|
case outbound
|
||||||
|
}
|
||||||
|
|
||||||
|
class Metrics {
|
||||||
|
var packetsNum: Int
|
||||||
|
var bytesNum: Int
|
||||||
|
|
||||||
|
init() {
|
||||||
|
self.packetsNum = 0
|
||||||
|
self.bytesNum = 0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private var inboundCounters: [String: Metrics] = [:]
|
||||||
|
private var outboundCounters: [String: Metrics] = [:]
|
||||||
|
|
||||||
|
private var printTask: Task<Void, Never>?
|
||||||
|
|
||||||
|
func start() {
|
||||||
|
self.printTask = Task {
|
||||||
|
while true {
|
||||||
|
do {
|
||||||
|
try Task.checkCancellation()
|
||||||
|
try await Task.sleep(for: .seconds(1))
|
||||||
|
for (from, metric) in inboundCounters {
|
||||||
|
SDLLogger.log("[SDLUDPCounter] inbound from: \(from), packet: \(metric.packetsNum), bytes: \(metric.bytesNum)")
|
||||||
|
}
|
||||||
|
self.inboundCounters.removeAll()
|
||||||
|
|
||||||
|
for (from, metric) in outboundCounters {
|
||||||
|
SDLLogger.log("[SDLUDPCounter] outbound from: \(from), packet: \(metric.packetsNum), bytes: \(metric.bytesNum)")
|
||||||
|
}
|
||||||
|
self.outboundCounters.removeAll()
|
||||||
|
} catch {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func increment(direction: Direction, from: String, bytes: Int) {
|
||||||
|
switch direction {
|
||||||
|
case .inbound:
|
||||||
|
let metric = inboundCounters[from, default: .init()]
|
||||||
|
metric.packetsNum += 1
|
||||||
|
metric.bytesNum += bytes
|
||||||
|
inboundCounters[from] = metric
|
||||||
|
case .outbound:
|
||||||
|
let metric = outboundCounters[from, default: .init()]
|
||||||
|
metric.packetsNum += 1
|
||||||
|
metric.bytesNum += bytes
|
||||||
|
outboundCounters[from] = metric
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func stop() {
|
||||||
|
self.printTask?.cancel()
|
||||||
|
self.printTask = nil
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@ -58,6 +58,9 @@ actor SDLUDPHole {
|
|||||||
self.udpHoleHandler.stop()
|
self.udpHoleHandler.stop()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
deinit {
|
||||||
|
SDLLogger.log("[SDLUDPHoleActor] deinit", for: .debug)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 处理和sn-server服务器之间的通讯
|
// 处理和sn-server服务器之间的通讯
|
||||||
@ -73,11 +76,15 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler {
|
|||||||
private let messageContinuation: AsyncThrowingStream<(SocketAddress, SDLHoleMessage), Error>.Continuation
|
private let messageContinuation: AsyncThrowingStream<(SocketAddress, SDLHoleMessage), Error>.Continuation
|
||||||
private var isMessageContinuationFinished: Bool = false
|
private var isMessageContinuationFinished: Bool = false
|
||||||
|
|
||||||
|
private var counterActor: SDLUDPCounter
|
||||||
|
|
||||||
// 启动函数
|
// 启动函数
|
||||||
init() throws {
|
init() throws {
|
||||||
let (stream, continuation) = AsyncThrowingStream.makeStream(of: (SocketAddress, SDLHoleMessage).self, bufferingPolicy: .bufferingNewest(2048))
|
let (stream, continuation) = AsyncThrowingStream.makeStream(of: (SocketAddress, SDLHoleMessage).self, bufferingPolicy: .bufferingNewest(2048))
|
||||||
self.messageStream = stream
|
self.messageStream = stream
|
||||||
self.messageContinuation = continuation
|
self.messageContinuation = continuation
|
||||||
|
|
||||||
|
self.counterActor = SDLUDPCounter()
|
||||||
}
|
}
|
||||||
|
|
||||||
func start() throws -> SocketAddress {
|
func start() throws -> SocketAddress {
|
||||||
@ -95,6 +102,10 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler {
|
|||||||
|
|
||||||
self.channel = channel
|
self.channel = channel
|
||||||
|
|
||||||
|
Task {
|
||||||
|
await self.counterActor.start()
|
||||||
|
}
|
||||||
|
|
||||||
return localAddress
|
return localAddress
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -105,7 +116,10 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler {
|
|||||||
var buffer = envelope.data
|
var buffer = envelope.data
|
||||||
let remoteAddress = envelope.remoteAddress
|
let remoteAddress = envelope.remoteAddress
|
||||||
|
|
||||||
SDLLogger.log("[SDLUDPHole] get raw bytes: \(buffer.readableBytes), from: \(remoteAddress)", for: .debug)
|
let bytesCount = buffer.readableBytes
|
||||||
|
Task {
|
||||||
|
await self.counterActor.increment(direction: .inbound, from: remoteAddress.description, bytes: bytesCount)
|
||||||
|
}
|
||||||
|
|
||||||
do {
|
do {
|
||||||
if let message = try SDLHoleMessage.decode(buffer: &buffer) {
|
if let message = try SDLHoleMessage.decode(buffer: &buffer) {
|
||||||
@ -119,12 +133,10 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func channelInactive(context: ChannelHandlerContext) {
|
func channelInactive(context: ChannelHandlerContext) {
|
||||||
SDLLogger.log("[SDLUDPHole] channelInactive", for: .debug)
|
|
||||||
self.finishMessageContinuationIfNeed(throwing: .closed)
|
self.finishMessageContinuationIfNeed(throwing: .closed)
|
||||||
}
|
}
|
||||||
|
|
||||||
func errorCaught(context: ChannelHandlerContext, error: any Error) {
|
func errorCaught(context: ChannelHandlerContext, error: any Error) {
|
||||||
SDLLogger.log("[SDLUDPHole] channel error: \(error)", for: .debug)
|
|
||||||
context.close(promise: nil)
|
context.close(promise: nil)
|
||||||
self.finishMessageContinuationIfNeed(throwing: .errorCaught)
|
self.finishMessageContinuationIfNeed(throwing: .errorCaught)
|
||||||
}
|
}
|
||||||
@ -135,6 +147,10 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Task {
|
||||||
|
await self.counterActor.increment(direction: .outbound, from: remoteAddress.description, bytes: data.count)
|
||||||
|
}
|
||||||
|
|
||||||
var buffer = channel.allocator.buffer(capacity: data.count + 1)
|
var buffer = channel.allocator.buffer(capacity: data.count + 1)
|
||||||
buffer.writeBytes([type.rawValue])
|
buffer.writeBytes([type.rawValue])
|
||||||
buffer.writeBytes(data)
|
buffer.writeBytes(data)
|
||||||
@ -152,11 +168,16 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func stop() {
|
func stop() {
|
||||||
SDLLogger.log("[SDLUDPHole] stop", for: .debug)
|
|
||||||
self.finishMessageContinuationIfNeed(throwing: nil)
|
self.finishMessageContinuationIfNeed(throwing: nil)
|
||||||
try? self.channel?.close().wait()
|
try? self.channel?.close().wait()
|
||||||
self.channel = nil
|
self.channel = nil
|
||||||
try? self.group.syncShutdownGracefully()
|
try? self.group.syncShutdownGracefully()
|
||||||
|
|
||||||
|
Task {
|
||||||
|
await self.counterActor.stop()
|
||||||
|
}
|
||||||
|
|
||||||
|
SDLLogger.log("[SDLUDPHole] stopped", for: .debug)
|
||||||
}
|
}
|
||||||
|
|
||||||
private func finishMessageContinuationIfNeed(throwing error: SDLUDPHoleError?) {
|
private func finishMessageContinuationIfNeed(throwing error: SDLUDPHoleError?) {
|
||||||
@ -173,4 +194,8 @@ private final class SDLUDPHoleHandler: ChannelInboundHandler {
|
|||||||
self.messageContinuation.finish(throwing: error)
|
self.messageContinuation.finish(throwing: error)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
deinit {
|
||||||
|
SDLLogger.log("[SDLUDPHole] deinit", for: .debug)
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user