fix outbound
This commit is contained in:
parent
83376b5513
commit
e6c79d7519
@ -10,6 +10,8 @@ import NetworkExtension
|
||||
import NIOCore
|
||||
|
||||
actor PacketOutboundActor {
|
||||
private typealias PacketReadResult = (packets: [Data], protocols: [NSNumber])?
|
||||
|
||||
private let provider: NEPacketTunnelProvider
|
||||
private let sessionManager: SessionManager
|
||||
private let arpServer: ArpServer
|
||||
@ -18,7 +20,8 @@ actor PacketOutboundActor {
|
||||
private let superServiceProxy: SDLSuperServiceProxy
|
||||
private let udpHoleServiceProxy: SDLUDPHoleServiceProxy
|
||||
private let flowTracer: SDLFlowTracer
|
||||
private var packetReaderService: SDLPacketReaderService?
|
||||
private var packetReaderTask: Task<Void, Never>?
|
||||
private var packetReaderGeneration: UInt64 = 0
|
||||
|
||||
private var networkAddress: SDLConfiguration.NetworkAddress
|
||||
private var identityId: UInt32
|
||||
@ -65,19 +68,47 @@ actor PacketOutboundActor {
|
||||
}
|
||||
|
||||
func startPacketReader() async {
|
||||
if self.packetReaderService == nil {
|
||||
self.packetReaderService = SDLPacketReaderService(provider: self.provider) { [weak self] event in
|
||||
await self?.handlePacketReaderEvent(event)
|
||||
guard self.packetReaderTask == nil else {
|
||||
return
|
||||
}
|
||||
|
||||
self.packetReaderGeneration &+= 1
|
||||
|
||||
let provider = self.provider
|
||||
let generation = self.packetReaderGeneration
|
||||
|
||||
self.packetReaderTask = Task(priority: .high) { [weak self] in
|
||||
while !Task.isCancelled {
|
||||
guard let batch = await Self.readPackets(from: provider) else {
|
||||
break
|
||||
}
|
||||
|
||||
if Task.isCancelled {
|
||||
break
|
||||
}
|
||||
|
||||
for (data, number) in zip(batch.packets, batch.protocols) where number.int32Value == 2 {
|
||||
if Task.isCancelled {
|
||||
break
|
||||
}
|
||||
|
||||
if let packet = IPPacket(data) {
|
||||
await self?.handleTunPacket(packet)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await self.packetReaderService?.start()
|
||||
await self?.finishPacketReader(generation: generation)
|
||||
SDLLogger.log("[PacketOutboundActor] packet reader task finished")
|
||||
}
|
||||
}
|
||||
|
||||
func stop() async {
|
||||
let packetReaderService = self.packetReaderService
|
||||
self.packetReaderService = nil
|
||||
await packetReaderService?.stop()
|
||||
self.packetReaderGeneration &+= 1
|
||||
|
||||
let packetReaderTask = self.packetReaderTask
|
||||
self.packetReaderTask = nil
|
||||
packetReaderTask?.cancel()
|
||||
}
|
||||
|
||||
func handleTunPacket(_ packet: IPPacket) async {
|
||||
@ -91,10 +122,26 @@ actor PacketOutboundActor {
|
||||
await self.handleTunRouteDecision(decision)
|
||||
}
|
||||
|
||||
private func handlePacketReaderEvent(_ event: SDLPacketReaderService.Event) async {
|
||||
switch event {
|
||||
case .packet(let packet):
|
||||
await self.handleTunPacket(packet)
|
||||
private func finishPacketReader(generation: UInt64) {
|
||||
guard generation == self.packetReaderGeneration else {
|
||||
return
|
||||
}
|
||||
|
||||
self.packetReaderTask = nil
|
||||
}
|
||||
|
||||
private static func readPackets(from provider: NEPacketTunnelProvider) async -> PacketReadResult {
|
||||
let readContinuation = OnceContinuation<PacketReadResult, Never>()
|
||||
|
||||
return await withTaskCancellationHandler {
|
||||
await withCheckedContinuation { continuation in
|
||||
readContinuation.set(continuation)
|
||||
provider.packetFlow.readPackets { packets, protocols in
|
||||
readContinuation.resume(returning: (packets: packets, protocols: protocols))
|
||||
}
|
||||
}
|
||||
} onCancel: {
|
||||
readContinuation.resume(returning: nil)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -1,98 +0,0 @@
|
||||
import Foundation
|
||||
import NetworkExtension
|
||||
|
||||
actor SDLPacketReaderService {
|
||||
enum Event {
|
||||
case packet(IPPacket)
|
||||
}
|
||||
|
||||
typealias EventHandler = @Sendable (Event) async -> Void
|
||||
private typealias PacketReadResult = (packets: [Data], protocols: [NSNumber])?
|
||||
|
||||
private final class CancellationToken: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var cancelled: Bool = false
|
||||
|
||||
var isCancelled: Bool {
|
||||
lock.lock()
|
||||
defer {
|
||||
lock.unlock()
|
||||
}
|
||||
return cancelled
|
||||
}
|
||||
|
||||
func cancel() {
|
||||
lock.lock()
|
||||
cancelled = true
|
||||
lock.unlock()
|
||||
}
|
||||
}
|
||||
|
||||
private let provider: NEPacketTunnelProvider
|
||||
private let onEvent: EventHandler
|
||||
private var readTask: Task<Void, Never>?
|
||||
private var readToken: CancellationToken?
|
||||
|
||||
init(provider: NEPacketTunnelProvider, onEvent: @escaping EventHandler) {
|
||||
self.provider = provider
|
||||
self.onEvent = onEvent
|
||||
}
|
||||
|
||||
func start() {
|
||||
guard self.readTask == nil else {
|
||||
return
|
||||
}
|
||||
|
||||
let provider = self.provider
|
||||
let onEvent = self.onEvent
|
||||
let readToken = CancellationToken()
|
||||
self.readToken = readToken
|
||||
|
||||
self.readTask = Task(priority: .high) {
|
||||
while !Task.isCancelled && !readToken.isCancelled {
|
||||
guard let batch = await Self.readPackets(from: provider) else {
|
||||
break
|
||||
}
|
||||
|
||||
if Task.isCancelled || readToken.isCancelled {
|
||||
break
|
||||
}
|
||||
|
||||
for (data, number) in zip(batch.packets, batch.protocols) where number.int32Value == 2 {
|
||||
if Task.isCancelled || readToken.isCancelled {
|
||||
break
|
||||
}
|
||||
|
||||
if let packet = IPPacket(data) {
|
||||
await onEvent(.packet(packet))
|
||||
}
|
||||
}
|
||||
}
|
||||
SDLLogger.log("[SDLPacketReaderService] readTask finished")
|
||||
}
|
||||
}
|
||||
|
||||
func stop() {
|
||||
let readTask = self.readTask
|
||||
self.readTask = nil
|
||||
let readToken = self.readToken
|
||||
self.readToken = nil
|
||||
readToken?.cancel()
|
||||
readTask?.cancel()
|
||||
}
|
||||
|
||||
private static func readPackets(from provider: NEPacketTunnelProvider) async -> PacketReadResult {
|
||||
let readContinuation = OnceContinuation<PacketReadResult, Never>()
|
||||
|
||||
return await withTaskCancellationHandler {
|
||||
await withCheckedContinuation { continuation in
|
||||
readContinuation.set(continuation)
|
||||
provider.packetFlow.readPackets { packets, protocols in
|
||||
readContinuation.resume(returning: (packets: packets, protocols: protocols))
|
||||
}
|
||||
}
|
||||
} onCancel: {
|
||||
readContinuation.resume(returning: nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user