diff --git a/Sources/SwiftNetwork/Protocols/NetworkEvents.swift b/Sources/SwiftNetwork/Protocols/NetworkEvents.swift index 484c032..f554d09 100644 --- a/Sources/SwiftNetwork/Protocols/NetworkEvents.swift +++ b/Sources/SwiftNetwork/Protocols/NetworkEvents.swift @@ -111,6 +111,7 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible { enum InternalEvent: Equatable { case dataStall case connectionIdle(idle: Bool) + case outboundDataPending(pending: Bool) case quic(event: QUICApplicationEvent) #if !NETWORK_EMBEDDED case custom(event: any DomainSpecificApplicationEvent) @@ -120,6 +121,7 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible { switch (lhs, rhs) { case (.dataStall, .dataStall): return true case (.connectionIdle(let lState), .connectionIdle(let rState)): return lState == rState + case (.outboundDataPending(let lState), .outboundDataPending(let rState)): return lState == rState default: return false } @@ -131,6 +133,7 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible { switch internalEvent { case .dataStall: return nil case .connectionIdle: return nil + case .outboundDataPending: return nil case .quic(let event): return event.domain #if !NETWORK_EMBEDDED case .custom(let event): return event.domain @@ -154,6 +157,14 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible { .init(internalEvent: .connectionIdle(idle: false)) } + static public var outboundDataBatchStart: ApplicationEvent { + .init(internalEvent: .outboundDataPending(pending: true)) + } + + static public var outboundDataBatchEnd: ApplicationEvent { + .init(internalEvent: .outboundDataPending(pending: false)) + } + public init(quicEvent: QUICApplicationEvent) { self.internalEvent = .quic(event: quicEvent) } @@ -173,6 +184,12 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible { } else { return "Connection Reused" } + case .outboundDataPending(let pending): + if pending { + return "Outbound Data Pending" + } else { + return "Outbound Data No Longer Pending" + } case .quic(let event): return event.description #if !NETWORK_EMBEDDED case .custom(let event): return event.description diff --git a/Sources/SwiftNetwork/QUIC/QUICConnection.swift b/Sources/SwiftNetwork/QUIC/QUICConnection.swift index 9c5dd7c..d63390c 100644 --- a/Sources/SwiftNetwork/QUIC/QUICConnection.swift +++ b/Sources/SwiftNetwork/QUIC/QUICConnection.swift @@ -359,6 +359,8 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol, private(set) var testSendingShortPackets = false private(set) var migrationSupported = false + private var pendOutboundData = false // Don't immediately process application sends + // false == IPv6, true == IPv4 private(set) var initialAddressIsIPv4 = false @@ -2460,6 +2462,11 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol, } // Note: trigger sending of any frames based on this external event checkConnectionIdle() + + guard !pendOutboundData else { + log.datapath("Outbound data pended, ignore send frames") + return + } sendFrames() } @@ -5490,6 +5497,11 @@ extension QUICConnection { // Note: trigger sendFrames() based on this external event checkConnectionIdle() + + guard !pendOutboundData else { + log.datapath("Outbound data pended, ignore send frames") + return + } if !sendFrames() { log.datapath("failed to send DATAGRAM frames") } @@ -5889,6 +5901,19 @@ extension QUICConnection { } public func handleApplicationEvent(_ event: ApplicationEvent) -> HandleNetworkEventResult { + if event == .outboundDataBatchStart { + // Start pending processing + pendOutboundData = true + return .consumed + } + + if event == .outboundDataBatchEnd { + // Stop pending processing, resume sending + pendOutboundData = false + sendFrames() + return .consumed + } + guard let quicEvent = event.quicEvent else { return .unconsumed } diff --git a/Tests/SwiftNetworkTests/QUICTestHarness.swift b/Tests/SwiftNetworkTests/QUICTestHarness.swift index 1e36733..43dc910 100644 --- a/Tests/SwiftNetworkTests/QUICTestHarness.swift +++ b/Tests/SwiftNetworkTests/QUICTestHarness.swift @@ -550,7 +550,8 @@ final class QUICTestHarness { dataGenerator: TestDataGenerator, streamIndex: Int, readChunkSize: Int = .max, - timeout: TimeInterval = 5.0 + timeout: TimeInterval = 5.0, + shouldBatchSends: Bool = false ) { guard let state else { XCTFail("State must be non-nil") @@ -585,6 +586,9 @@ final class QUICTestHarness { // Write on the client stream context.async { var chunkCount = 1 + if shouldBatchSends { + state.clientHarness.invokeApplicationEvent(.outboundDataBatchStart) + } for dataChunk in dataGenerator { var writeResult = false if dataGenerator.numberOfBlocks == chunkCount && dataGenerator.sendFIN { @@ -595,6 +599,9 @@ final class QUICTestHarness { XCTAssertTrue(writeResult) chunkCount += 1 } + if shouldBatchSends { + state.clientHarness.invokeApplicationEvent(.outboundDataBatchEnd) + } } // Wait for the server stream @@ -910,6 +917,7 @@ final class QUICTestHarness { sendStreamStopSendingError: Bool = false, verifyResetStreamHalfClosure: Bool = false, shouldMarkIdle: Bool = false, + shouldBatchSends: Bool = false, clientOptions: ProtocolOptions = QUICProtocol.options(), serverOptions: ProtocolOptions = QUICProtocol.options(), sendMaxStreamUpdate: Bool = false, @@ -1010,7 +1018,8 @@ final class QUICTestHarness { dataGenerator: generator, streamIndex: index, readChunkSize: clientReadChunkSize, - timeout: timeout + timeout: timeout, + shouldBatchSends: shouldBatchSends ) } else { XCTAssertTrue(dataBlock == nil && blockSize == 0 && blockCount == 0) diff --git a/Tests/SwiftNetworkTests/SwiftNetworkQUICHarnessTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkQUICHarnessTests.swift index 6b470b3..911959b 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkQUICHarnessTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkQUICHarnessTests.swift @@ -389,6 +389,10 @@ final class SwiftNetworkQUICHarnessTests: NetTestCase { QUICTestHarness().runQUICTest(blockSize: 10240, blockCount: 4) } + func testQUICEcho40KiBBatched() { + QUICTestHarness().runQUICTest(blockSize: 10240, blockCount: 4, shouldBatchSends: true) + } + func testQUICEcho40KiBSmallReads() { QUICTestHarness().runQUICTest(blockSize: 10240, blockCount: 4, clientReadChunkSize: 1000) }