Merge pull request #53 from apple/agnosticdev/InboundStopping

SwiftQUIC: Reduce CPU overhead in getting inbound stream data delivered to the stream

Author: agnosticdev

Sources/SwiftNetwork/Protocols/ManyToManyProtocol.swift

@@ -1388,6 +1388,15 @@ extension ManyToManyApplicationStreamProtocol where Flow: AutomaticUpperStreamPr
         guard let flow = self.flow(for: flowID) else { throw NetworkError.posix(EINVAL) }
         flow.serviceUpperReceiveQueue()
     }
+    // Enqueue and delivery the stream data all in one shot
+    public func deliverInboundStreamData(
+        flow flowID: MultiplexedFlowIdentifier,
+        streamData: consuming FrameArray
+    ) throws(NetworkError) {
+        guard var flow = self.flow(for: flowID) else { throw NetworkError.posix(EINVAL) }
+        try flow.addToUpperReceiveQueue(streamData)
+        flow.serviceUpperReceiveQueue()
+    }
 }
 
 @_spi(ProtocolProvider)
@@ -1581,6 +1590,16 @@ extension ManyToManyApplicationDatagramProtocol where Flow: AutomaticUpperDatagr
         guard let flow = self.flow(for: flowID) else { throw NetworkError.posix(EINVAL) }
         flow.serviceUpperReceiveQueue()
     }
+
+    // Enqueue and delivery the datagrams all in one shot
+    public func deliverInboundDatagrams(
+        flow flowID: MultiplexedFlowIdentifier,
+        datagrams: consuming FrameArray
+    ) throws(NetworkError) {
+        guard var flow = self.flow(for: flowID) else { throw NetworkError.posix(EINVAL) }
+        try flow.addToUpperReceiveQueue(datagrams)
+        flow.serviceUpperReceiveQueue()
+    }
 }
 
 @available(Network 0.1.0, *)
@@ -1613,6 +1632,16 @@ extension HeterogeneousManyToManyProtocolHandler where SecondaryFlow: AutomaticU
         guard let flow = self.secondaryFlow(for: flowID) else { throw NetworkError.posix(EINVAL) }
         flow.serviceUpperReceiveQueue()
     }
+
+    // Enqueue and delivery the datagrams all in one shot
+    public func deliverInboundDatagrams(
+        flow flowID: MultiplexedFlowIdentifier,
+        datagrams: consuming FrameArray
+    ) throws(NetworkError) {
+        guard var flow = self.secondaryFlow(for: flowID) else { throw NetworkError.posix(EINVAL) }
+        try flow.addToUpperReceiveQueue(datagrams)
+        flow.serviceUpperReceiveQueue()
+    }
 }
 
 @_spi(ProtocolProvider)

Sources/SwiftNetwork/QUIC/QUICConnection.swift

@@ -2374,8 +2374,7 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol,
                 continue
             }
 
-            try? enqueueInboundStreamData(flow: flowID, streamData: frameArray)
-            try? deliverEnqueuedInboundStreamData(flow: flowID)
+            try? deliverInboundStreamData(flow: flowID, streamData: frameArray)
 
             // When the stream is already in `resetReceived` state,
             // it should be closed when we receive STOP_SENDING, so we
@@ -4317,8 +4316,7 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol,
         }
         if let frameArray = stream.dequeueReassembledData(connection: self) {
             do {
-                try enqueueInboundStreamData(flow: flowID, streamData: frameArray)
-                try deliverEnqueuedInboundStreamData(flow: flowID)
+                try deliverInboundStreamData(flow: flowID, streamData: frameArray)
                 sendFrames()
             } catch {
                 log.error("Error sending frames on stream close: \(error)")
@@ -5564,8 +5562,7 @@ extension QUICConnection {
 
         var frame = frame.frame
         frame.metadataComplete = true
-        try? enqueueInboundDatagrams(flow: matchingFlowIdentifier, datagrams: .init(frame: frame))
-        try? deliverEnqueuedInboundDatagrams(flow: matchingFlowIdentifier)
+        try? deliverInboundDatagrams(flow: matchingFlowIdentifier, datagrams: .init(frame: frame))
         return true
     }
 }

Sources/SwiftNetwork/QUIC/QUICStream.swift

@@ -328,7 +328,7 @@ struct StreamListMembership: OptionSet {
 // Also note that a flow identifier can exist in multiple lists at one time.
 @available(Network 0.1.0, *)
 struct QUICStreamList: ~Copyable {
-    private var list: [MultiplexedFlowIdentifier] = []
+    private var list = Deque<MultiplexedFlowIdentifier>()
     private let name: StaticString
     private let listType: StreamListMembership
 

Tests/SwiftNetworkTests/TestMultiplexingProtocol.swift

@@ -91,8 +91,7 @@ final class TestMultiplexingProtocol: ManyToManyApplicationDatagramProtocol, Man
             return
         }
         accessReceivedDatagrams(path: path) { frames in
-            try? enqueueInboundDatagrams(flow: flow, datagrams: frames.drainArray())
-            try? deliverEnqueuedInboundDatagrams(flow: flow)
+            try? deliverInboundDatagrams(flow: flow, datagrams: frames.drainArray())
         }
     }