Mark Sendable for NetworkContext's external async calls

Author: tfpauly

Sources/SwiftNetwork/Connection/Connection.swift

@@ -2085,7 +2085,7 @@ extension NetworkChannel where ApplicationProtocol: MessageProtocol {
 
 @_spi(Essentials)
 @available(Network 0.1.0, *)
-extension NetworkChannel where ApplicationProtocol: StreamProtocol {
+extension NetworkChannel where ApplicationProtocol: StreamProtocol & Sendable {
     public struct StreamMessage {
         public static func message(content: [UInt8]? = nil, isComplete: Bool = false) -> StreamMessage {
             StreamMessage(content: content, isComplete: isComplete)
@@ -2097,10 +2097,12 @@ extension NetworkChannel where ApplicationProtocol: StreamProtocol {
 
     public func send(_ message: StreamMessage, completion: (@Sendable (Result<Void, NetworkError>) -> Void)? = nil) {
         let endpointFlow = self.endpointFlow
+        let content = message.content
+        let isComplete = message.isComplete
         endpointFlow.async {
             let writeRequest = WriteRequest(
-                content: message.content,
-                isComplete: message.isComplete,
+                content: content,
+                isComplete: isComplete,
                 completion: completion
             )
             endpointFlow.addWriteRequestOnContext(writeRequest)
@@ -2114,10 +2116,12 @@ extension NetworkChannel where ApplicationProtocol: StreamProtocol {
         completion: (@Sendable (Result<Void, NetworkError>) -> Void)? = nil
     ) {
         let endpointFlow = self.endpointFlow
+        nonisolated(unsafe) let unsafeBuffer = buffer
+        nonisolated(unsafe) let unsafeOwner = owner
         endpointFlow.async {
             let writeRequest = WriteRequest(
-                buffer: buffer,
-                owner: owner,
+                buffer: unsafeBuffer,
+                owner: unsafeOwner,
                 isComplete: isComplete,
                 completion: completion
             )
@@ -2144,7 +2148,7 @@ extension NetworkChannel where ApplicationProtocol: StreamProtocol {
 
 @_spi(Essentials)
 @available(Network 0.1.0, *)
-extension NetworkChannel where ApplicationProtocol: DatagramProtocol {
+extension NetworkChannel where ApplicationProtocol: DatagramProtocol & Sendable {
     public struct DatagramMessage {
         public static func message(content: [UInt8]? = nil) -> DatagramMessage {
             DatagramMessage(content: content)
@@ -2155,8 +2159,9 @@ extension NetworkChannel where ApplicationProtocol: DatagramProtocol {
 
     public func send(_ message: DatagramMessage, completion: (@Sendable (Result<Void, NetworkError>) -> Void)? = nil) {
         let endpointFlow = self.endpointFlow
+        let content = message.content
         endpointFlow.async {
-            let writeRequest = WriteRequest(content: message.content, isComplete: true, completion: completion)
+            let writeRequest = WriteRequest(content: content, isComplete: true, completion: completion)
             endpointFlow.addWriteRequestOnContext(writeRequest)
         }
     }
@@ -2168,10 +2173,12 @@ extension NetworkChannel where ApplicationProtocol: DatagramProtocol {
         completion: (@Sendable (Result<Void, NetworkError>) -> Void)? = nil
     ) {
         let endpointFlow = self.endpointFlow
+        nonisolated(unsafe) let unsafeBuffer = buffer
+        nonisolated(unsafe) let unsafeOwner = owner
         endpointFlow.async {
             let writeRequest = WriteRequest(
-                buffer: buffer,
-                owner: owner,
+                buffer: unsafeBuffer,
+                owner: unsafeOwner,
                 isComplete: isComplete,
                 completion: completion
             )

Sources/SwiftNetwork/Context/NetworkContext.swift

@@ -46,8 +46,9 @@ public protocol NetworkContextProtocol: AnyObject, Hashable {
     init(identifier: String)
     var identifier: String { get }
     func activate()
-    func async(_ block: @escaping () -> Void)
-    func barrierAsync(_ block: @escaping () -> Void)
+    func async(_ block: @Sendable @escaping () -> Void)
+    func isolatedAsync(_ block: @escaping () -> Void)
+    func barrierAsync(_ block: @Sendable @escaping () -> Void)
 }
 
 @_spi(Essentials)
@@ -65,12 +66,17 @@ public final class NetworkContext: NetworkContextProtocol, @unchecked Sendable {
     }
 
     public protocol Scheduler: AnyObject {
+        /// Runs an immediate task, scheduled from the scheduler's context (so the completion is non-sendable).
+        /// No assumptions are made about how the task is run.
+        func runImmediateIsolated(_ task: @escaping () -> Void)
+
         /// Runs an immediate task. No assumptions are made about how the task is run.
-        func runImmediate(_ task: @escaping (() -> Void))
+        func runImmediate(_ task: @Sendable @escaping () -> Void)
+
         /// Schedules a task to run after a delay, using a reference.
         ///
         /// The `milliseconds` parameter specifies the delay before the task runs.
-        func schedule(_ task: @escaping (() -> Void), milliseconds: Int64, reference: TimerReference)
+        func schedule(_ task: @escaping () -> Void, milliseconds: Int64, reference: TimerReference)
         /// Unschedules a task with a reference.
         func unschedule(reference: TimerReference)
         /// A Boolean value that indicates whether the current code is running in the scheduler.
@@ -301,7 +307,7 @@ extension NetworkContext {
                 entries.removeFirst(where: { $0.reference == reference })
             }
 
-            func insert(targetTime: DispatchTime, reference: TimerReference, task: @escaping (() -> Void)) {
+            func insert(targetTime: DispatchTime, reference: TimerReference, task: @escaping () -> Void) {
                 // First, remove an existing entry for this reference
                 remove(by: reference)
                 // Check to see if the targetTime is before the first entry before it's added
@@ -348,14 +354,19 @@ extension NetworkContext {
         init(globals: Globals) {
             self.globals = globals
         }
+        /// Runs an immediate task, scheduled from the scheduler's context (so the completion is non-sendable).
+        /// No assumptions are made about how the task is run.
+        func runImmediateIsolated(_ task: @escaping () -> Void) {
+            globals.queue.async(execute: DispatchWorkItem(block: task))
+        }
         /// Runs an immediate task. No assumptions are made about how the task is run.
-        func runImmediate(_ task: @escaping (() -> Void)) {
+        func runImmediate(_ task: @Sendable @escaping () -> Void) {
             globals.queue.async(execute: DispatchWorkItem(block: task))
         }
         /// Schedules a task to run after a delay, using a reference.
         ///
         /// The `milliseconds` parameter specifies the delay before the task runs.
-        func schedule(_ task: @escaping (() -> Void), milliseconds: Int64, reference: TimerReference) {
+        func schedule(_ task: @escaping () -> Void, milliseconds: Int64, reference: TimerReference) {
             let targetTime = DispatchTime.now() + DispatchTimeInterval.milliseconds(Int(milliseconds))
             globals.timerList.insert(targetTime: targetTime, reference: reference, task: task)
         }
@@ -380,11 +391,18 @@ extension NetworkContext {
         globals.queue
     }
 
-    public func async(_ block: @escaping () -> Void) {
+    public func async(_ block: @Sendable @escaping () -> Void) {
         scheduler.runImmediate(block)
     }
 
-    public func barrierAsync(_ block: @escaping () -> Void) {
+    public func isolatedAsync(_ block: @escaping () -> Void) {
+#if DEBUG
+        assert()
+#endif
+        scheduler.runImmediateIsolated(block)
+    }
+
+    public func barrierAsync(_ block: @Sendable @escaping () -> Void) {
         scheduler.runImmediate(block)
     }
 

Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift

@@ -32,7 +32,7 @@ internal import Synchronization
 #endif
 
 @available(Network 0.1.0, *)
-final class EndpointFlow: CustomDebugStringConvertible {
+final class EndpointFlow: CustomDebugStringConvertible, @unchecked Sendable {
 
     /// State used to emit logs on the data path.
     public var log = NetworkLoggerState()
@@ -269,7 +269,7 @@ final class EndpointFlow: CustomDebugStringConvertible {
         }
     }
 
-    func async(_ block: @escaping () -> Void) {
+    func async(_ block: @Sendable @escaping () -> Void) {
         self.parameters.context.async(block)
     }
 

Sources/SwiftNetwork/EndpointFlow/ReadRequest.swift

@@ -13,11 +13,11 @@
 //===----------------------------------------------------------------------===//
 
 @available(Network 0.1.0, *)
-struct ReadRequest {
+struct ReadRequest: Sendable {
     let minimumBytes: Int
     let maximumBytes: Int
     let maximumFrames: Int
-    let completion: ([UInt8]?, Bool, Bool, NetworkError?) -> Void
+    let completion: @Sendable ([UInt8]?, Bool, Bool, NetworkError?) -> Void
 
     func complete(content: [UInt8]?, isComplete: Bool, isFinal: Bool, error: NetworkError? = nil) {
         completion(content, isComplete, isFinal, error)

Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift

@@ -520,7 +520,7 @@ extension NetworkContext {
 
     fileprivate func async(index: NetworkStateIndex, _ block: @escaping () -> Void) {
         self.softAssert()
-        self.async {
+        self.isolatedAsync {
             self.protocolEventStates[index].startAsyncCall()
             defer {
                 self.protocolEventStates[index].finishAsyncCall()

Sources/SwiftNetwork/QUIC/QUICConnection.swift

@@ -5253,7 +5253,7 @@ extension QUICConnection {
         }
         asyncSendRunning = true
         log.datapath("async: scheduling restart after packet burst")
-        self.context.async {
+        self.async {
             self.resumeSendingAfterBurstLimit()
         }
     }

Sources/SwiftNetworkBenchmarks/BenchmarkUtility.swift

@@ -72,7 +72,7 @@ public struct QUICLoopbackState {
 
 @_spi(ProtocolProvider)
 @available(Network 0.1.0, *)
-public struct QUICClientEndpointResult {
+public struct QUICClientEndpointResult: @unchecked Sendable {
     public var instance: QUICConnection
     public var parameters: Parameters
     public var upperHandler: StreamUpperHarness
@@ -82,7 +82,7 @@ public struct QUICClientEndpointResult {
 
 @_spi(ProtocolProvider)
 @available(Network 0.1.0, *)
-public struct QUICServerEndpointResult {
+public struct QUICServerEndpointResult: @unchecked Sendable {
     public var instance: QUICConnection
     public var parameters: Parameters
     public var upperHandler: NewStreamFlowHarness
@@ -96,12 +96,12 @@ public enum BenchmarkError: Error {
 
 @_spi(ProtocolProvider)
 @available(Network 0.1.0, *)
-public final class QUICBenchmarkUtility {
+public final class QUICBenchmarkUtility: Sendable {
 
     // 127.0.0.1 (Just point both at loopback)
     public static let localIPv4Address: [UInt8] = [0x7f, 0x00, 0x00, 0x01]
     public static let remoteIPv4Address: [UInt8] = [0x7f, 0x00, 0x00, 0x01]
-    var serverSigningKey = P256.Signing.PrivateKey()
+    let serverSigningKey = P256.Signing.PrivateKey()
 
     public func createQUICTestOptions(
         server: Bool = false,
@@ -265,20 +265,20 @@ public final class QUICBenchmarkUtility {
 #if !(os(Linux) || NETWORK_EMBEDDED)
 @available(macOS 11, iOS 14, tvOS 14, watchOS 7, *)
 #endif
-public struct LoggingHandle: CustomStringConvertible {
+public struct LoggingHandle: CustomStringConvertible, Sendable {
     #if os(Linux) || NETWORK_EMBEDDED
     let logger = Logger(label: "com.apple.network.benchmarks")
     #else
     #if canImport(os)
     let logger = Logger(subsystem: "com.apple.network.benchmarks", category: "perf")
     #endif
     #endif
-    public enum LoggingType: Int {
+    public enum LoggingType: Int, Sendable {
         case none = 0
         case print = 1
         case log = 2
     }
-    public var loggingType: LoggingType = .none
+    public let loggingType: LoggingType
 
     public init(loggingType: LoggingType) {
         self.loggingType = loggingType
@@ -322,7 +322,7 @@ public struct LoggingHandle: CustomStringConvertible {
 
 @_spi(ProtocolProvider)
 @available(Network 0.1.0, *)
-public final class DataBenchmarkUtility {
+public final class DataBenchmarkUtility: Sendable {
     @discardableResult
     public func loopOutputHandlerPackets(
         sender: DatagramLowerHarness,

Sources/Tools/IPUDPTransfer/main.swift

@@ -27,7 +27,7 @@ internal import os
 #endif
 
 @available(Network 0.1.0, *)
-final class IPUDPTransfer {
+final class IPUDPTransfer: @unchecked Sendable {
 
     // 169.254.156.146
     let localIPv4Address: [UInt8] = [0xa9, 0xfe, 0x9c, 0x92]
@@ -37,14 +37,14 @@ final class IPUDPTransfer {
     let dataBenchmarkUtility = DataBenchmarkUtility()
     let NSEC_PER_MSEC = UInt64(Duration.milliseconds(1) / Duration.nanoseconds(1))
 
+    var iterationIndex = 0
+
     func run(iterations: Int, packets: Int, loggingHandle: LoggingHandle, group: DispatchGroup, sendSize: Int) -> Double
     {
-        let ipv4Client = Endpoint(address: IPv4Address(localIPv4Address)!, port: 0)
-        let ipv4Server = Endpoint(address: IPv4Address(remoteIPv4Address)!, port: 0)
         // Create a random payload to send back and forth
-        var payload = [UInt8](repeating: 0, count: sendSize)
-        payload = (0..<sendSize).map { _ in UInt8.random(in: 0...255) }
-        var iterationIndex = 0
+        var mutablePayload = [UInt8](repeating: 0, count: sendSize)
+        mutablePayload = (0..<sendSize).map { _ in UInt8.random(in: 0...255) }
+        let payload = mutablePayload
         print(
             "Running IP/UDP transfer, transferring \(iterations) iteration\(iterations > 1 ? "s" : "") of \(packets) packet\(packets > 1 ? "s" : "")"
         )
@@ -53,11 +53,15 @@ final class IPUDPTransfer {
         // NOTE: Today this benchmark relies on DispatchGroups due to Sendability issues on some of the project types.
         // In the future we should try to adopt Swift Concurrency.
         group.enter()
-        var clientParameters = Parameters()
         let context = NetworkContext(identifier: "IPUDPTransfer")
-        clientParameters.context = context
         context.activate()
         context.async {
+            let ipv4Client = Endpoint(address: IPv4Address(self.localIPv4Address)!, port: 0)
+            let ipv4Server = Endpoint(address: IPv4Address(self.remoteIPv4Address)!, port: 0)
+
+            var clientParameters = Parameters()
+            clientParameters.context = context
+
             for _ in 0..<iterations {
                 // Client
                 let path = PathProperties(parameters: clientParameters)
@@ -185,7 +189,7 @@ final class IPUDPTransfer {
                 serverInput.stop()
                 serverInput.teardown()
 
-                iterationIndex += 1
+                self.iterationIndex += 1
             }
             group.leave()
         }

Sources/Tools/QUICHandshake/main.swift

@@ -29,7 +29,7 @@ internal import os
 #if IMPORT_SWIFTTLS && canImport(SwiftTLS)
 
 @available(Network 0.1.0, *)
-final class QUICHandshake {
+final class QUICHandshake: @unchecked Sendable {
 
     let quicBenchmarkUtility = QUICBenchmarkUtility()
     let dataBenchmarkUtility = DataBenchmarkUtility()
@@ -137,7 +137,7 @@ final class QUICHandshake {
                     }
                 }
                 // Wait until both instances are connected to signal handshake complete
-                func pollForConnectedInstances() {
+                @Sendable func pollForConnectedInstances() {
                     // Wait for the client connection state to go into connected
                     if client.instance.state == .connected {
                         loggingHandle.log("Client connection handshake finished")