Compare commits

..

8 Commits

Author SHA1 Message Date
6cb8641869 fix task group 2025-08-03 12:26:54 +08:00
2d6adba8a8 fix 2025-08-03 01:31:38 +08:00
c8f1787ea2 fix 2025-08-03 01:13:52 +08:00
95760764a5 fix 2025-08-03 01:04:53 +08:00
5756422ebb fix 2025-08-03 00:24:53 +08:00
e6fc903439 add debug 2025-08-03 00:16:08 +08:00
d775f5ba96 add debug 2025-08-03 00:10:32 +08:00
a292bec2c4 add log 2025-08-02 23:56:40 +08:00
3 changed files with 30 additions and 9 deletions

View File

@ -120,7 +120,7 @@ public class SDLContext: @unchecked Sendable {
do { do {
try await self.startSuperClient() try await self.startSuperClient()
} catch let err { } catch let err {
NSLog("SuperClient get error: \(err)") NSLog("[SDLContext] SuperClient get error: \(err), will restart")
await self.arpServer.clear() await self.arpServer.clear()
try? await Task.sleep(for: .seconds(2)) try? await Task.sleep(for: .seconds(2))
} }

View File

@ -9,8 +9,6 @@ import Foundation
import NIOCore import NIOCore
import NIOPosix import NIOPosix
// --MARK: SuperNode // --MARK: SuperNode
actor SDLSuperClient { actor SDLSuperClient {
private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
@ -69,6 +67,10 @@ actor SDLSuperClient {
} }
group.addTask { group.addTask {
defer {
self.inboundContinuation.finish()
}
for try await var packet in inbound { for try await var packet in inbound {
if let message = SDLSuperClientDecoder.decode(buffer: &packet) { if let message = SDLSuperClientDecoder.decode(buffer: &packet) {
SDLLogger.log("[SDLSuperTransport] read message: \(message)", level: .warning) SDLLogger.log("[SDLSuperTransport] read message: \(message)", level: .warning)
@ -82,9 +84,14 @@ actor SDLSuperClient {
} }
} }
} }
NSLog("[SDLSuperClient] inbound closed")
} }
group.addTask { group.addTask {
defer {
self.writeContinuation.finish()
}
for try await message in self.writeStream { for try await message in self.writeStream {
var buffer = self.asyncChannel.channel.allocator.buffer(capacity: message.data.count + 5) var buffer = self.asyncChannel.channel.allocator.buffer(capacity: message.data.count + 5)
buffer.writeInteger(message.packetId, as: UInt32.self) buffer.writeInteger(message.packetId, as: UInt32.self)
@ -92,17 +99,28 @@ actor SDLSuperClient {
buffer.writeBytes(message.data) buffer.writeBytes(message.data)
try await outbound.write(buffer) try await outbound.write(buffer)
} }
NSLog("[SDLSuperClient] outbound closed")
} }
// --MARK: // --MARK:
group.addTask { group.addTask {
while !Task.isCancelled { while true {
await self.ping() do {
try? await Task.sleep(nanoseconds: 5 * 1_000_000_000) await self.ping()
try await Task.sleep(nanoseconds: 5 * 1_000_000_000)
} catch let err {
NSLog("[SDLSuperClient] heartbeat cancelled with error: \(err)")
break
}
} }
} }
try await group.waitForAll() // 退,
for try await _ in group {
}
NSLog("[SDLSuperClient] group closed")
} }
} }
} }
@ -182,7 +200,6 @@ actor SDLSuperClient {
deinit { deinit {
try! group.syncShutdownGracefully() try! group.syncShutdownGracefully()
self.inboundContinuation.finish()
} }
} }

View File

@ -100,7 +100,11 @@ actor SDLUDPHole {
} }
} }
try await group.waitForAll() for try await _ in group {
}
SDLLogger.log("[SDLUDPHole] group closed", level: .debug)
} }
} }
} }