Skip to content

Mark Sendable for NetworkContext's external async calls #36

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tfpauly wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Mark Sendable for NetworkContext's external async calls #36

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 24 additions & 17 deletions Sources/SwiftNetwork/Connection/Connection.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -969,7 +969,7 @@ public struct TCP: StreamProtocol {
#if !NETWORK_EMBEDDED
@_spi(Essentials)
@available(Network 0.1.0, *)
public struct DatagramBridge: DatagramProtocol {
public struct DatagramBridge: DatagramProtocol, Sendable {
public typealias ContentType = Void

public let belowProtocol: Void
Expand All @@ -990,7 +990,7 @@ public struct DatagramBridge: DatagramProtocol {
#if !NETWORK_EMBEDDED
@_spi(Essentials)
@available(Network 0.1.0, *)
public struct StreamBridge: StreamProtocol {
public struct StreamBridge: StreamProtocol, Sendable {
public typealias ContentType = Void

public let belowProtocol: Void
Expand All @@ -1010,8 +1010,8 @@ public struct StreamBridge: StreamProtocol {

@_spi(Essentials)
@available(Network 0.1.0, *)
public struct NoTransport: StreamProtocol {
public enum BelowProtocol {
public struct NoTransport: StreamProtocol, Sendable {
public enum BelowProtocol: Sendable {
case void
#if !NETWORK_EMBEDDED
case bridge(StreamBridge)
Expand Down Expand Up @@ -1331,7 +1331,7 @@ public struct QUIC: MultiplexProtocol {
/// > Note: This type is not intended to be inserted into the protocol stack manually; it is vended by connections that use QUIC.
@_spi(Essentials)
@available(Network 0.1.0, *)
public struct QUICStream: StreamProtocol {
public struct QUICStream: StreamProtocol, Sendable {
public let belowProtocol: Void

public enum Directionality: Equatable {
Expand All @@ -1353,7 +1353,7 @@ public struct QUICStream: StreamProtocol {
/// UDP supports sending and receiving datagrams.
@_spi(Essentials)
@available(Network 0.1.0, *)
public struct UDP: DatagramProtocol {
public struct UDP: DatagramProtocol, Sendable {
public typealias ContentType = [UInt8]

private var preferNoChecksum: Bool?
Expand Down Expand Up @@ -1397,8 +1397,8 @@ public struct UDP: DatagramProtocol {
/// protocol stack when configuring IP options.
@_spi(Essentials)
@available(Network 0.1.0, *)
public struct IP: NetworkProtocolOptions {
public enum BelowProtocol {
public struct IP: NetworkProtocolOptions, Sendable {
public enum BelowProtocol: Sendable {
case void
#if !NETWORK_EMBEDDED
case bridge(DatagramBridge)
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand All @@ -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

@agnosticdev agnosticdev Jun 30, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are we able to build the WriteRequest and then pass it into the block to avoid the nonisolated(unsafe) ?

endpointFlow.async {
let writeRequest = WriteRequest(
buffer: buffer,
owner: owner,
buffer: unsafeBuffer,
owner: unsafeOwner,
isComplete: isComplete,
completion: completion
)
Expand All @@ -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)
Expand All @@ -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)
}
}
Expand All @@ -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
)
Expand Down
36 changes: 27 additions & 9 deletions Sources/SwiftNetwork/Context/NetworkContext.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
}
Expand All @@ -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)
}

Expand Down
4 changes: 2 additions & 2 deletions Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ internal import Synchronization
#endif

@available(Network 0.1.0, *)
final class EndpointFlow: CustomDebugStringConvertible {
final class EndpointFlow: CustomDebugStringConvertible, @unchecked Sendable {

@glbrntt glbrntt Jun 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In NIO we have a rule that any use of @unchecked Sendable must include the rationale for why they are @unchecked. It's been incredibly valuable. Our long term goal is to remove all uses of @unchecked. You may find it helpful to adopt a similar pattern here.

Using @unchcked is often a footgun: it's easy to add new properties which aren't Sendable but might, for example, require being on the context to access safely. In NIO we usually isolate any state like that to its own struct within the type and only mark that as @unchecked but include runtime checks (for debug builds) to make sure we aren't violating those contracts. It was actually one such check that led us to this PR.


/// State used to emit logs on the data path.
public var log = NetworkLoggerState()
Expand Down Expand Up @@ -269,7 +269,7 @@ final class EndpointFlow: CustomDebugStringConvertible {
}
}

func async(_ block: @escaping () -> Void) {
func async(_ block: @Sendable @escaping () -> Void) {
self.parameters.context.async(block)
}

Expand Down
4 changes: 2 additions & 2 deletions Sources/SwiftNetwork/EndpointFlow/ReadRequest.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
4 changes: 2 additions & 2 deletions Sources/SwiftNetwork/Protocols/IPProtocol.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ public struct IPProtocol: NetworkProtocol {
// Destination Address: UInt128
}

public enum Version: UInt8 {
public enum Version: UInt8, Sendable {
/// Allows any IP version.
case any = 0
/// Uses only IP version 4 (IPv4).
Expand All @@ -64,7 +64,7 @@ public struct IPProtocol: NetworkProtocol {
case v6 = 6
}

public enum AddressPreference: UInt8 {
public enum AddressPreference: UInt8, Sendable {
case any = 0
case temporary = 1
case stable = 2
Expand Down
2 changes: 1 addition & 1 deletion Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
4 changes: 2 additions & 2 deletions Sources/SwiftNetwork/QUIC/QUICConnection.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4023,7 +4023,7 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol,
else {
return
}
context.async {
self.async {
var flowType: QLogFlowType = .client
var applicationType = "client"
if self.isServer {
Expand Down Expand Up @@ -5253,7 +5253,7 @@ extension QUICConnection {
}
asyncSendRunning = true
log.datapath("async: scheduling restart after packet burst")
self.context.async {
self.async {
self.resumeSendingAfterBurstLimit()
}
}
Expand Down
16 changes: 8 additions & 8 deletions Sources/SwiftNetworkBenchmarks/BenchmarkUtility.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading