fix
This commit is contained in:
parent
e40a266b13
commit
d4390f5117
@ -13,7 +13,7 @@ actor SDLPuncherActor {
|
|||||||
private var coolingDown: Set<Data> = []
|
private var coolingDown: Set<Data> = []
|
||||||
private let cooldown: Duration = .seconds(5)
|
private let cooldown: Duration = .seconds(5)
|
||||||
|
|
||||||
private var udpHole: SDLUDPHole?
|
private var udpHole: SDLUDPHole
|
||||||
|
|
||||||
private var pktId: UInt32 = 1
|
private var pktId: UInt32 = 1
|
||||||
// 提交后还没有响应的请求
|
// 提交后还没有响应的请求
|
||||||
@ -23,22 +23,31 @@ actor SDLPuncherActor {
|
|||||||
private var logger: SDLLogger
|
private var logger: SDLLogger
|
||||||
private var querySocketAddress: SocketAddress
|
private var querySocketAddress: SocketAddress
|
||||||
|
|
||||||
|
private let(requestStream, requestContinuation) = AsyncStream.makeStream(of: RegisterRequest.self, bufferingPolicy: .unbounded)
|
||||||
|
|
||||||
struct RegisterRequest {
|
struct RegisterRequest {
|
||||||
let srcMac: Data
|
let srcMac: Data
|
||||||
let dstMac: Data
|
let dstMac: Data
|
||||||
let networkId: UInt32
|
let networkId: UInt32
|
||||||
}
|
}
|
||||||
|
|
||||||
init(querySocketAddress: SocketAddress, logger: SDLLogger) {
|
init(udpHole: SDLUDPHole, querySocketAddress: SocketAddress, logger: SDLLogger) {
|
||||||
|
self.udpHole = udpHole
|
||||||
self.querySocketAddress = querySocketAddress
|
self.querySocketAddress = querySocketAddress
|
||||||
self.logger = logger
|
self.logger = logger
|
||||||
}
|
}
|
||||||
|
|
||||||
func setUDPHoleActor(udpHole: SDLUDPHole?) {
|
nonisolated func submitRegisterRequest(request: RegisterRequest) {
|
||||||
self.udpHole = udpHole
|
self.requestContinuation.yield(request)
|
||||||
}
|
}
|
||||||
|
|
||||||
func submitRegisterRequest(request: RegisterRequest) {
|
func startConsuming() async {
|
||||||
|
for await request in self.requestStream {
|
||||||
|
self.submitRegisterRequest(request: request)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private func handleRegisterRequest(request: RegisterRequest) {
|
||||||
let dstMac = request.dstMac
|
let dstMac = request.dstMac
|
||||||
|
|
||||||
guard !coolingDown.contains(dstMac) else {
|
guard !coolingDown.contains(dstMac) else {
|
||||||
@ -73,7 +82,7 @@ actor SDLPuncherActor {
|
|||||||
register.srcMac = request.srcMac
|
register.srcMac = request.srcMac
|
||||||
register.dstMac = request.dstMac
|
register.dstMac = request.dstMac
|
||||||
|
|
||||||
self.udpHole?.send(type: .register, data: try! register.serializedData(), remoteAddress: remoteAddress)
|
self.udpHole.send(type: .register, data: try! register.serializedData(), remoteAddress: remoteAddress)
|
||||||
} else {
|
} else {
|
||||||
self.logger.log("[SDLContext] hole sock address is invalid: \(peerInfo.v4Info)", level: .warning)
|
self.logger.log("[SDLContext] hole sock address is invalid: \(peerInfo.v4Info)", level: .warning)
|
||||||
}
|
}
|
||||||
@ -95,7 +104,7 @@ actor SDLPuncherActor {
|
|||||||
self.pendingRequests[pktId] = request
|
self.pendingRequests[pktId] = request
|
||||||
|
|
||||||
if let queryData = try? queryInfo.serializedData() {
|
if let queryData = try? queryInfo.serializedData() {
|
||||||
self.udpHole?.send(type: .queryInfo, data: queryData, remoteAddress: self.querySocketAddress)
|
self.udpHole.send(type: .queryInfo, data: queryData, remoteAddress: self.querySocketAddress)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -45,7 +45,7 @@ public class SDLContext {
|
|||||||
// 依赖的变量
|
// 依赖的变量
|
||||||
var udpHole: SDLUDPHole?
|
var udpHole: SDLUDPHole?
|
||||||
var providerAdapter: SDLTunnelProviderAdapter
|
var providerAdapter: SDLTunnelProviderAdapter
|
||||||
var puncherActor: SDLPuncherActor
|
var puncherActor: SDLPuncherActor?
|
||||||
// dns的client对象
|
// dns的client对象
|
||||||
var dnsClient: SDLDNSClient?
|
var dnsClient: SDLDNSClient?
|
||||||
|
|
||||||
@ -68,7 +68,7 @@ public class SDLContext {
|
|||||||
private var noticeClient: SDLNoticeClient?
|
private var noticeClient: SDLNoticeClient?
|
||||||
|
|
||||||
// 流量统计
|
// 流量统计
|
||||||
private var flowTracer = SDLFlowTracerActor()
|
private var flowTracer = SDLFlowTracer()
|
||||||
private var flowTracerCancel: AnyCancellable?
|
private var flowTracerCancel: AnyCancellable?
|
||||||
|
|
||||||
private let logger: SDLLogger
|
private let logger: SDLLogger
|
||||||
@ -83,7 +83,6 @@ public class SDLContext {
|
|||||||
self.sessionManager = SessionManager()
|
self.sessionManager = SessionManager()
|
||||||
self.arpServer = ArpServer(known_macs: [:])
|
self.arpServer = ArpServer(known_macs: [:])
|
||||||
self.providerAdapter = SDLTunnelProviderAdapter(provider: provider, logger: logger)
|
self.providerAdapter = SDLTunnelProviderAdapter(provider: provider, logger: logger)
|
||||||
self.puncherActor = SDLPuncherActor(querySocketAddress: config.stunSocketAddress, logger: logger)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public func start() async throws {
|
public func start() async throws {
|
||||||
@ -160,7 +159,7 @@ public class SDLContext {
|
|||||||
case .registerSuperNak(let registerSuperNak):
|
case .registerSuperNak(let registerSuperNak):
|
||||||
await self.handleRegisterSuperNak(nakPacket: registerSuperNak)
|
await self.handleRegisterSuperNak(nakPacket: registerSuperNak)
|
||||||
case .peerInfo(let peerInfo):
|
case .peerInfo(let peerInfo):
|
||||||
await self.puncherActor.handlePeerInfo(peerInfo: peerInfo)
|
await self.puncherActor?.handlePeerInfo(peerInfo: peerInfo)
|
||||||
case .event(let event):
|
case .event(let event):
|
||||||
try await self.handleEvent(event: event)
|
try await self.handleEvent(event: event)
|
||||||
case .stunProbeReply(let probeReply):
|
case .stunProbeReply(let probeReply):
|
||||||
@ -214,8 +213,10 @@ public class SDLContext {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private func handleUDPHoleReady() async throws {
|
private func handleUDPHoleReady() async throws {
|
||||||
await self.puncherActor.setUDPHoleActor(udpHole: self.udpHole)
|
if let udpHole = self.udpHole {
|
||||||
|
self.puncherActor = SDLPuncherActor(udpHole: udpHole, querySocketAddress: config.stunSocketAddress, logger: logger)
|
||||||
|
}
|
||||||
|
|
||||||
await withTaskGroup(of: Void.self) { group in
|
await withTaskGroup(of: Void.self) { group in
|
||||||
group.addTask {
|
group.addTask {
|
||||||
// 开始探测nat的类型
|
// 开始探测nat的类型
|
||||||
@ -369,7 +370,7 @@ public class SDLContext {
|
|||||||
|
|
||||||
let layerPacket = try LayerPacket(layerData: decyptedData)
|
let layerPacket = try LayerPacket(layerData: decyptedData)
|
||||||
|
|
||||||
await self.flowTracer.inc(num: decyptedData.count, type: .inbound)
|
self.flowTracer.inc(num: decyptedData.count, type: .inbound)
|
||||||
// 处理arp请求
|
// 处理arp请求
|
||||||
switch layerPacket.type {
|
switch layerPacket.type {
|
||||||
case .arp:
|
case .arp:
|
||||||
@ -507,15 +508,15 @@ public class SDLContext {
|
|||||||
if let session = self.sessionManager.getSession(toAddress: dstMac) {
|
if let session = self.sessionManager.getSession(toAddress: dstMac) {
|
||||||
self.logger.log("[SDLContext] send packet by session: \(session)", level: .debug)
|
self.logger.log("[SDLContext] send packet by session: \(session)", level: .debug)
|
||||||
self.udpHole?.send(type: .data, data: data, remoteAddress: session.natAddress)
|
self.udpHole?.send(type: .data, data: data, remoteAddress: session.natAddress)
|
||||||
await self.flowTracer.inc(num: data.count, type: .p2p)
|
self.flowTracer.inc(num: data.count, type: .p2p)
|
||||||
}
|
}
|
||||||
else {
|
else {
|
||||||
// 通过super_node进行转发
|
// 通过super_node进行转发
|
||||||
self.udpHole?.send(type: .data, data: data, remoteAddress: self.config.stunSocketAddress)
|
self.udpHole?.send(type: .data, data: data, remoteAddress: self.config.stunSocketAddress)
|
||||||
// 流量统计
|
// 流量统计
|
||||||
await self.flowTracer.inc(num: data.count, type: .forward)
|
self.flowTracer.inc(num: data.count, type: .forward)
|
||||||
// 尝试打洞
|
// 尝试打洞
|
||||||
await self.puncherActor.submitRegisterRequest(request: .init(srcMac: networkAddr.mac, dstMac: dstMac, networkId: networkAddr.networkId))
|
self.puncherActor?.submitRegisterRequest(request: .init(srcMac: networkAddr.mac, dstMac: dstMac, networkId: networkAddr.networkId))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -6,9 +6,10 @@
|
|||||||
//
|
//
|
||||||
|
|
||||||
import Foundation
|
import Foundation
|
||||||
|
import Darwin
|
||||||
|
|
||||||
// 流量统计器
|
// 流量统计器
|
||||||
actor SDLFlowTracerActor {
|
final class SDLFlowTracer {
|
||||||
enum FlowType {
|
enum FlowType {
|
||||||
case forward
|
case forward
|
||||||
case p2p
|
case p2p
|
||||||
@ -19,7 +20,14 @@ actor SDLFlowTracerActor {
|
|||||||
private var p2pFlowBytes: UInt32 = 0
|
private var p2pFlowBytes: UInt32 = 0
|
||||||
private var inFlowBytes: UInt32 = 0
|
private var inFlowBytes: UInt32 = 0
|
||||||
|
|
||||||
|
private let lock = NSLock()
|
||||||
|
|
||||||
func inc(num: Int, type: FlowType) {
|
func inc(num: Int, type: FlowType) {
|
||||||
|
lock.lock()
|
||||||
|
defer {
|
||||||
|
lock.unlock()
|
||||||
|
}
|
||||||
|
|
||||||
switch type {
|
switch type {
|
||||||
case .inbound:
|
case .inbound:
|
||||||
self.inFlowBytes += UInt32(num)
|
self.inFlowBytes += UInt32(num)
|
||||||
@ -31,13 +39,14 @@ actor SDLFlowTracerActor {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func reset() -> (UInt32, UInt32, UInt32) {
|
func reset() -> (UInt32, UInt32, UInt32) {
|
||||||
|
lock.lock()
|
||||||
defer {
|
defer {
|
||||||
self.forwardFlowBytes = 0
|
self.forwardFlowBytes = 0
|
||||||
self.inFlowBytes = 0
|
self.inFlowBytes = 0
|
||||||
self.p2pFlowBytes = 0
|
self.p2pFlowBytes = 0
|
||||||
|
lock.unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
return (forwardFlowBytes, p2pFlowBytes, inFlowBytes)
|
return (forwardFlowBytes, p2pFlowBytes, inFlowBytes)
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
Loading…
x
Reference in New Issue
Block a user