From 8ad78b46375d64e2d2d03f6aca4a64610f842630 Mon Sep 17 00:00:00 2001 From: agnosticdev Date: Sat, 8 Aug 2026 13:48:06 -0700 Subject: [PATCH 1/4] SwiftQUIC: New inbound streams should enqueue the inboundDataAvailable event --- .../Protocols/ProtocolDatagramHandlers.swift | 11 ++- .../Protocols/ProtocolStreamHandlers.swift | 11 ++- Sources/Tools/QUICTransfer/main.swift | 83 ++++++++++--------- 3 files changed, 64 insertions(+), 41 deletions(-) diff --git a/Sources/SwiftNetwork/Protocols/ProtocolDatagramHandlers.swift b/Sources/SwiftNetwork/Protocols/ProtocolDatagramHandlers.swift index 5d334d6..443369c 100644 --- a/Sources/SwiftNetwork/Protocols/ProtocolDatagramHandlers.swift +++ b/Sources/SwiftNetwork/Protocols/ProtocolDatagramHandlers.swift @@ -121,7 +121,16 @@ extension AutomaticUpperDatagramProcessing where Self: ~Copyable { /// Notifies the upper protocol that frames are available in `upperReceiveQueue`. public func serviceUpperReceiveQueue() { guard !upperReceiveQueue.isEmpty else { return } - upper.deliverInboundDataAvailableEvent(reference) + if upper.isDetached { + // Enqueue pending event until the daragram is completely attached. + // This will ensure that the inboundDataAvailable event is sent for all new datagrams carrying data + let selfReference = self.reference + selfReference.enqueuePendingEventForUpperProtocol( + event: .inboundDataAvailable(selfReference, upper.reference) + ) + } else { + upper.deliverInboundDataAvailableEvent(reference) + } } } diff --git a/Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift b/Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift index 2fde116..2ab7016 100644 --- a/Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift +++ b/Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift @@ -133,7 +133,16 @@ extension AutomaticUpperStreamProcessing where Self: ~Copyable { /// Notifies the upper protocol that frames are available in `upperReceiveQueue`. public func serviceUpperReceiveQueue() { guard !upperReceiveQueue.isEmpty else { return } - upper.deliverInboundDataAvailableEvent(reference) + if upper.isDetached { + // Enqueue pending event until the stream is completely attached. + // This will ensure that the inboundDataAvailable event is sent for all new streams carrying data + let selfReference = self.reference + selfReference.enqueuePendingEventForUpperProtocol( + event: .inboundDataAvailable(selfReference, upper.reference) + ) + } else { + upper.deliverInboundDataAvailableEvent(reference) + } } } diff --git a/Sources/Tools/QUICTransfer/main.swift b/Sources/Tools/QUICTransfer/main.swift index 6e7c47a..9ae73cf 100644 --- a/Sources/Tools/QUICTransfer/main.swift +++ b/Sources/Tools/QUICTransfer/main.swift @@ -62,7 +62,6 @@ final class QUICTransfer { // Create a random payload to send back and forth var payload = [UInt8](repeating: 0, count: sendSize) payload = (0.. 1 ? "s" : "")") let timestart = DispatchTime.now().uptimeNanoseconds @@ -251,53 +250,59 @@ final class QUICTransfer { } var serverStream: StreamUpperHarness? + var writeIndex = 0 + var writeSucceeded = true var totalReadSize = 0 - while index < iterations { - group.enter() + let totalExpectedSize = iterations * payload.count + let doneSemaphore = DispatchSemaphore(value: 0) + + // Client write loop to perform all writes until finished + func writeLoop() { + guard writeIndex < iterations else { return } + guard clientStream.write(payload) else { + loggingHandle.log("Client failed to write at iteration: \(writeIndex)") + writeSucceeded = false + return + } + writeIndex += 1 context.async { - guard clientStream.write(payload) else { - loggingHandle.log("Client failed to write at iteration: \(index)") - group.leave() - return - } - if serverStream == nil { - group.enter() - serverInput.waitForNewFlow { - loggingHandle.log("Server got new inbound flow") - serverStream = serverInput.upperHarnesses.last - group.leave() - } - } - group.leave() + writeLoop() } - group.wait() + } - guard let serverStream else { - return 0 + // Server read loop: keeps draining inbound data as it arrives. + // Readloop used for multiple iterations + func readLoop(stream: StreamUpperHarness) { + stream.waitForInboundDataAvailable { available in + guard available else { return } + totalReadSize += stream.readAndDrop() + if totalReadSize >= totalExpectedSize { + doneSemaphore.signal() + } else { + readLoop(stream: stream) + } } + } - group.enter() - context.async { - var serverReadDataSizeForIteration = 0 - var serverReadCompletion: ((Bool) -> Void)? = nil - serverReadCompletion = { _ in - let readBytes = serverStream.readAndDrop() - if readBytes > 0 { - serverReadDataSizeForIteration += readBytes - totalReadSize += readBytes - } - if serverReadDataSizeForIteration >= payload.count { - serverReadCompletion = nil - index += 1 - group.leave() - } else { - serverStream.waitForInboundDataAvailable(completion: serverReadCompletion!) - } + context.async { + // Setup inbound flow observer and then start the client write loop + serverInput.waitForNewFlow { + serverStream = serverInput.upperHarnesses.last + if let serverStream { + readLoop(stream: serverStream) + } else { + doneSemaphore.signal() } - serverStream.waitForInboundDataAvailable(completion: serverReadCompletion!) } - group.wait() + writeLoop() } + doneSemaphore.wait() + + guard serverStream != nil, writeSucceeded else { + return 0 + } + + let index = min(totalReadSize / payload.count, iterations) group.enter() context.async { From 77b43eff2a6fc061e324065c5d6dcd233e484204 Mon Sep 17 00:00:00 2001 From: agnosticdev Date: Mon, 10 Aug 2026 06:34:44 -0700 Subject: [PATCH 2/4] Reverted to just call read on the stream --- .../Protocols/ProtocolDatagramHandlers.swift | 11 +---------- .../SwiftNetwork/Protocols/ProtocolEventManager.swift | 1 + Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift | 2 ++ .../Protocols/ProtocolStreamHandlers.swift | 11 +---------- Sources/Tools/QUICTransfer/main.swift | 9 +++++++++ 5 files changed, 14 insertions(+), 20 deletions(-) diff --git a/Sources/SwiftNetwork/Protocols/ProtocolDatagramHandlers.swift b/Sources/SwiftNetwork/Protocols/ProtocolDatagramHandlers.swift index 443369c..5d334d6 100644 --- a/Sources/SwiftNetwork/Protocols/ProtocolDatagramHandlers.swift +++ b/Sources/SwiftNetwork/Protocols/ProtocolDatagramHandlers.swift @@ -121,16 +121,7 @@ extension AutomaticUpperDatagramProcessing where Self: ~Copyable { /// Notifies the upper protocol that frames are available in `upperReceiveQueue`. public func serviceUpperReceiveQueue() { guard !upperReceiveQueue.isEmpty else { return } - if upper.isDetached { - // Enqueue pending event until the daragram is completely attached. - // This will ensure that the inboundDataAvailable event is sent for all new datagrams carrying data - let selfReference = self.reference - selfReference.enqueuePendingEventForUpperProtocol( - event: .inboundDataAvailable(selfReference, upper.reference) - ) - } else { - upper.deliverInboundDataAvailableEvent(reference) - } + upper.deliverInboundDataAvailableEvent(reference) } } diff --git a/Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift b/Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift index aa6cd90..1f3e4f5 100644 --- a/Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift +++ b/Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift @@ -73,6 +73,7 @@ struct ProtocolEventManagerState: ~Copyable { enum PendingEvent: ~Copyable { case connected(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference) case disconnected(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference, error: NetworkError?) + // inboundDataAvailable event will only be available on the second inbound datagram, the first will not trigger this event case inboundDataAvailable(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference) case outboundRoomAvailable(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference) case inboundAborted(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference, error: NetworkError?) diff --git a/Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift b/Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift index 085ff0a..7d54e1b 100644 --- a/Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift +++ b/Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift @@ -64,6 +64,8 @@ extension UpperProtocolLinkage { @_spi(ProtocolProvider) @available(Network 0.1.0, *) public protocol InboundDataLinkage: UpperProtocolLinkage where PairedLinkage: OutboundDataLinkage { + // Will only notify a upper protocol for the second inbound datagram. + // To receive data for the first inbound datagram read directly on the upper protocol. func deliverInboundDataAvailableEvent(_ from: ProtocolInstanceReference) func deliverOutboundRoomAvailableEvent(_ from: ProtocolInstanceReference) } diff --git a/Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift b/Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift index 2ab7016..2fde116 100644 --- a/Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift +++ b/Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift @@ -133,16 +133,7 @@ extension AutomaticUpperStreamProcessing where Self: ~Copyable { /// Notifies the upper protocol that frames are available in `upperReceiveQueue`. public func serviceUpperReceiveQueue() { guard !upperReceiveQueue.isEmpty else { return } - if upper.isDetached { - // Enqueue pending event until the stream is completely attached. - // This will ensure that the inboundDataAvailable event is sent for all new streams carrying data - let selfReference = self.reference - selfReference.enqueuePendingEventForUpperProtocol( - event: .inboundDataAvailable(selfReference, upper.reference) - ) - } else { - upper.deliverInboundDataAvailableEvent(reference) - } + upper.deliverInboundDataAvailableEvent(reference) } } diff --git a/Sources/Tools/QUICTransfer/main.swift b/Sources/Tools/QUICTransfer/main.swift index 9ae73cf..04e28eb 100644 --- a/Sources/Tools/QUICTransfer/main.swift +++ b/Sources/Tools/QUICTransfer/main.swift @@ -289,6 +289,15 @@ final class QUICTransfer { serverInput.waitForNewFlow { serverStream = serverInput.upperHarnesses.last if let serverStream { + // If there is only one inbound read then a read can take place here and that is it. + // If there are more data after the first read then a read loop will need to be setup + // to observe the rest of the inbound data events. + totalReadSize += serverStream.readAndDrop() + if totalReadSize >= totalExpectedSize { + doneSemaphore.signal() + } else { + readLoop(stream: serverStream) + } readLoop(stream: serverStream) } else { doneSemaphore.signal() From cc729283d7760040ed3994bca20477d81c7e1edd Mon Sep 17 00:00:00 2001 From: agnosticdev Date: Tue, 11 Aug 2026 13:58:46 -0700 Subject: [PATCH 3/4] QUIC Only flag --- Sources/Tools/QUICTransfer/main.swift | 151 +++++++++++++++++--------- 1 file changed, 99 insertions(+), 52 deletions(-) diff --git a/Sources/Tools/QUICTransfer/main.swift b/Sources/Tools/QUICTransfer/main.swift index 04e28eb..86446a8 100644 --- a/Sources/Tools/QUICTransfer/main.swift +++ b/Sources/Tools/QUICTransfer/main.swift @@ -52,7 +52,8 @@ final class QUICTransfer { loggingHandle: LoggingHandle, group: DispatchGroup, sendSize: Int, - linkDelay: NetworkDuration = .zero + linkDelay: NetworkDuration = .zero, + quicOnly: Bool = false ) -> Double { let ipv4Client = Endpoint(address: IPv4Address(localIPv4Address)!, port: 1234) let ipv4Server = Endpoint(address: IPv4Address(remoteIPv4Address)!, port: 2345) @@ -82,14 +83,18 @@ final class QUICTransfer { let clientIPOptions = IPProtocol.options() clientIPOptions.setLogID(prefix: "C", parent: "1", protocolLogIDNumber: 3) clientIPOptions.setProtocolInstance(clientIP) - clientParameters.defaultStack.internet = .ip(clientIPOptions) + if !quicOnly { + clientParameters.defaultStack.internet = .ip(clientIPOptions) + } let clientUDP = UDPProtocol.instance(context: context) let clientUDPOptions = UDPProtocol.options() clientUDPOptions.noMetadata = true clientUDPOptions.setLogID(prefix: "C", parent: "1", protocolLogIDNumber: 2) clientUDPOptions.setProtocolInstance(clientUDP) - clientParameters.defaultStack.transport = .udp(clientUDPOptions) + if !quicOnly { + clientParameters.defaultStack.transport = .udp(clientUDPOptions) + } let clientQUIC = QUICProtocol.instance(context: context) var clientTLSOptions = SwiftTLSProtocol.Options() @@ -102,7 +107,11 @@ final class QUICTransfer { clientQUICOptions.tlsOptions = clientTLSOptions clientQUICOptions.setLogID(prefix: "C", parent: "1", protocolLogIDNumber: 1) clientQUICOptions.setProtocolInstance(clientQUIC) - clientParameters.defaultStack.prepend(applicationProtocol: .quic(clientQUICOptions)) + if !quicOnly { + clientParameters.defaultStack.prepend(applicationProtocol: .quic(clientQUICOptions)) + } else { + clientParameters.defaultStack.transport = .quic(clientQUICOptions) + } let clientOutput = BridgeDatagramProtocol.instance(context: clientParameters.context) let bridgeOptions = BridgeDatagramProtocol.options() @@ -135,27 +144,39 @@ final class QUICTransfer { return } do { - try clientQUIC.attachLowerDatagramProtocolForNewPath( - clientUDP, - remote: ipv4Server, - local: ipv4Client, - parameters: clientParameters, - path: path - ) - try clientUDP.attachLowerDatagramProtocol( - clientIP, - remote: ipv4Server, - local: ipv4Client, - parameters: clientParameters, - path: path - ) - try clientIP.attachLowerDatagramProtocol( - clientOutput, - remote: ipv4Server, - local: ipv4Client, - parameters: clientParameters, - path: path - ) + if !quicOnly { + // Use QUIC -> UDP -> IP -> BridgeProtocol + try clientQUIC.attachLowerDatagramProtocolForNewPath( + clientUDP, + remote: ipv4Server, + local: ipv4Client, + parameters: clientParameters, + path: path + ) + try clientUDP.attachLowerDatagramProtocol( + clientIP, + remote: ipv4Server, + local: ipv4Client, + parameters: clientParameters, + path: path + ) + try clientIP.attachLowerDatagramProtocol( + clientOutput, + remote: ipv4Server, + local: ipv4Client, + parameters: clientParameters, + path: path + ) + } else { + // Use only QUIC -> Bridge Protocol + try clientQUIC.attachLowerDatagramProtocolForNewPath( + clientOutput, + remote: ipv4Server, + local: ipv4Client, + parameters: clientParameters, + path: path + ) + } } catch { loggingHandle.log("Failed to attach client IP to lower protocol") group.leave() @@ -167,14 +188,18 @@ final class QUICTransfer { let serverIPOptions = IPProtocol.options() serverIPOptions.setLogID(prefix: "L", parent: "1", protocolLogIDNumber: 3) clientIPOptions.setProtocolInstance(serverIP) - serverParameters.defaultStack.internet = .ip(serverIPOptions) + if !quicOnly { + serverParameters.defaultStack.internet = .ip(serverIPOptions) + } let serverUDP = UDPProtocol.instance(context: context) let serverUDPOptions = UDPProtocol.options() serverUDPOptions.noMetadata = true serverUDPOptions.setLogID(prefix: "L", parent: "1", protocolLogIDNumber: 2) serverUDPOptions.setProtocolInstance(serverUDP) - serverParameters.defaultStack.transport = .udp(serverUDPOptions) + if !quicOnly { + serverParameters.defaultStack.transport = .udp(serverUDPOptions) + } let serverQUIC = QUICProtocol.instance(context: context) var serverTLSOptions = SwiftTLSProtocol.Options() @@ -186,7 +211,11 @@ final class QUICTransfer { serverQUICOptions.tlsOptions = serverTLSOptions serverQUICOptions.setLogID(prefix: "L", parent: "1", protocolLogIDNumber: 1) serverQUICOptions.setProtocolInstance(serverQUIC) - serverParameters.defaultStack.prepend(applicationProtocol: .quic(serverQUICOptions)) + if !quicOnly { + serverParameters.defaultStack.prepend(applicationProtocol: .quic(serverQUICOptions)) + } else { + serverParameters.defaultStack.transport = .quic(serverQUICOptions) + } let serverOutput = BridgeDatagramProtocol.instance(context: context) let serverBridgeOptions = BridgeDatagramProtocol.options() @@ -210,28 +239,40 @@ final class QUICTransfer { return } do { - try serverQUIC.attachLowerDatagramProtocolForNewPath( - serverUDP, - remote: ipv4Client, - local: ipv4Server, - parameters: serverParameters, - path: serverPath - ) - - try serverUDP.attachLowerDatagramProtocol( - serverIP, - remote: ipv4Client, - local: ipv4Server, - parameters: clientParameters, - path: path - ) - try serverIP.attachLowerDatagramProtocol( - serverOutput, - remote: ipv4Client, - local: ipv4Server, - parameters: serverParameters, - path: serverPath - ) + if !quicOnly { + // Use QUIC -> UDP -> IP -> BridgeProtocol + try serverQUIC.attachLowerDatagramProtocolForNewPath( + serverUDP, + remote: ipv4Client, + local: ipv4Server, + parameters: serverParameters, + path: serverPath + ) + + try serverUDP.attachLowerDatagramProtocol( + serverIP, + remote: ipv4Client, + local: ipv4Server, + parameters: clientParameters, + path: path + ) + try serverIP.attachLowerDatagramProtocol( + serverOutput, + remote: ipv4Client, + local: ipv4Server, + parameters: serverParameters, + path: serverPath + ) + } else { + // Use only QUIC -> Bridge Protocol + try serverQUIC.attachLowerDatagramProtocolForNewPath( + serverOutput, + remote: ipv4Client, + local: ipv4Server, + parameters: serverParameters, + path: serverPath + ) + } } catch { loggingHandle.log("Failed to attach server IP to lower protocol") group.leave() @@ -339,10 +380,11 @@ final class QUICTransfer { if #available(anyAppleOS 26, *) { // Take command line arguments - var iterations = 10000 // 5gb total (if 500000 sendSize) + var iterations = 1 // 5gb total (if 500000 sendSize) var loggingHandler: LoggingHandle = LoggingHandle(loggingType: .none) var sendSize = 500000 // 500kb var linkDelay = NetworkDuration.zero + var quicOnly = false let arguments = CommandLine.arguments.dropFirst(0) if arguments.contains("-iterations"), let index = arguments.firstIndex(of: "-iterations") @@ -354,6 +396,10 @@ if #available(anyAppleOS 26, *) { } } + if arguments.contains("-quicOnly") { + quicOnly = true + } + if arguments.contains("-logging"), let index = arguments.firstIndex(of: "-logging") { @@ -392,7 +438,8 @@ if #available(anyAppleOS 26, *) { loggingHandle: loggingHandler, group: group, sendSize: sendSize, - linkDelay: linkDelay + linkDelay: linkDelay, + quicOnly: quicOnly ) if totalTime > 0 { print("Finished all (\(iterations)) transfers in \(totalTime) seconds") From 6d43fc6f4aaf070041a0b78bada162c1c093b9be Mon Sep 17 00:00:00 2001 From: agnosticdev Date: Tue, 11 Aug 2026 14:13:41 -0700 Subject: [PATCH 4/4] Remove comments --- Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift | 1 - Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift | 2 -- 2 files changed, 3 deletions(-) diff --git a/Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift b/Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift index 1f3e4f5..aa6cd90 100644 --- a/Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift +++ b/Sources/SwiftNetwork/Protocols/ProtocolEventManager.swift @@ -73,7 +73,6 @@ struct ProtocolEventManagerState: ~Copyable { enum PendingEvent: ~Copyable { case connected(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference) case disconnected(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference, error: NetworkError?) - // inboundDataAvailable event will only be available on the second inbound datagram, the first will not trigger this event case inboundDataAvailable(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference) case outboundRoomAvailable(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference) case inboundAborted(_ from: ProtocolInstanceReference, _ to: ProtocolInstanceReference, error: NetworkError?) diff --git a/Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift b/Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift index 7d54e1b..085ff0a 100644 --- a/Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift +++ b/Sources/SwiftNetwork/Protocols/ProtocolLinkage.swift @@ -64,8 +64,6 @@ extension UpperProtocolLinkage { @_spi(ProtocolProvider) @available(Network 0.1.0, *) public protocol InboundDataLinkage: UpperProtocolLinkage where PairedLinkage: OutboundDataLinkage { - // Will only notify a upper protocol for the second inbound datagram. - // To receive data for the first inbound datagram read directly on the upper protocol. func deliverInboundDataAvailableEvent(_ from: ProtocolInstanceReference) func deliverOutboundRoomAvailableEvent(_ from: ProtocolInstanceReference) }