Reuse the received frame deque

Sources/SwiftNetwork/QUIC/Packet.swift

@@ -16,7 +16,6 @@
 
 #if canImport(BasicContainers)
 import BasicContainers
-internal import DequeModule
 #endif
 
 #if canImport(Glibc)
@@ -231,9 +230,6 @@ struct Packet: ~Copyable {
         identifier.space
     }
 
-    // Temporary storage for parsed frames
-    var framesReceived = NetworkUniqueDeque<QUICFrame>()
-
     // Temporary storage used for logging, only set when datapath logs or QLog is enabled
     var shorthandFrames: [QUICShorthandFrame]?
 
@@ -249,24 +245,6 @@ struct Packet: ~Copyable {
     // Version Negotiation
     private(set) var versions: [QUICVersion]?
 
-    mutating func cleanupReceivedFrames() {
-        while let receivedFrame = framesReceived.popFirst() {
-            let type = receivedFrame.frameType
-            switch type {
-            case .crypto:
-                guard case .crypto(var frame) = receivedFrame else { continue }
-                frame.frame.finalize(success: false)
-            case .stream:
-                guard case .stream(var frame) = receivedFrame else { continue }
-                frame.frame.finalize(success: false)
-            case .datagram:
-                guard case .datagram(var frame) = receivedFrame else { continue }
-                frame.frame.finalize(success: false)
-            default: continue
-            }
-        }
-    }
-
     // Flags
     struct Flags: OptionSet {
         init(rawValue: Self.RawValue) {

Sources/SwiftNetwork/QUIC/PacketParser.swift

@@ -33,8 +33,38 @@ internal import os
 struct PacketParser: ~Copyable, PrefixedLoggable {
     var log: LogPrefixer
 
+    // Temporary storage for the frames parsed out of the packet currently being
+    // processed.
+    var framesReceived = NetworkUniqueDeque<QUICFrame>()
+
+    // Initial capacity for `framesReceived`.
+    private static var framesReceivedCapacity: Int { 4 }
+
     init(logPrefixer: LogPrefixer) {
         self.log = logPrefixer
+        self.framesReceived.reserveCapacity(Self.framesReceivedCapacity)
+    }
+
+    // Drain any frames which haven't been processed, finalizing the ones holding
+    // borrowed buffers so that they aren't leaked.
+    mutating func cleanupReceivedFrames() {
+        while let receivedFrame = self.framesReceived.popFirst() {
+            switch receivedFrame {
+            case .crypto(var frame):
+                frame.frame.finalize(success: false)
+            case .stream(var frame):
+                frame.frame.finalize(success: false)
+            case .datagram(var frame):
+                frame.frame.finalize(success: false)
+            default: continue
+            }
+        }
+
+        // A packet stuffed with single byte frames can grow the deque a long way; a peer
+        // shouldn't be able to pin that storage for the lifetime of the connection.
+        if self.framesReceived.capacity > QUICPreferences.shared.maxReceivedFramesCapacity {
+            self.framesReceived.reallocate(capacity: Self.framesReceivedCapacity)
+        }
     }
 
     private func parsePacketNumber(
@@ -55,7 +85,7 @@ struct PacketParser: ~Copyable, PrefixedLoggable {
     }
 
     @inline(never)
-    private func parseFrames(
+    private mutating func parseFrames(
         frame: inout Frame,
         packet: inout Packet,
         connection: QUICConnection,
@@ -94,7 +124,7 @@ struct PacketParser: ~Copyable, PrefixedLoggable {
                 connection: connection,
                 isLastPacketInFrame: isLastPacketInFrame
             )
-            packet.framesReceived.append(quicFrame)
+            self.framesReceived.append(quicFrame)
         }
     }
 
@@ -158,13 +188,12 @@ struct PacketParser: ~Copyable, PrefixedLoggable {
         return reservedBits
     }
 
-    func parse(
+    mutating func parse(
         frame: inout Frame,
         connection: QUICConnection,
         path: QUICPath,
         ecn: IPProtocol.ECN
     ) -> Packet? {
-
         if _slowPath(frame.unclaimedLength < Constants.minimumPacketSize) {
             connection.log.error("Dropping short packet, len=\(frame.unclaimedLength)")
             return nil
@@ -293,21 +322,7 @@ struct PacketParser: ~Copyable, PrefixedLoggable {
                 )
             } catch {
                 // Explicitly release finalize frames in case of error
-                while let frame = packet.framesReceived.popFirst() {
-                    let frameType = frame.frameType
-                    switch frameType {
-                    case .crypto:
-                        guard case .crypto(var frame) = frame else { continue }
-                        frame.frame.finalize(success: false)
-                    case .stream:
-                        guard case .stream(var frame) = frame else { continue }
-                        frame.frame.finalize(success: false)
-                    case .datagram:
-                        guard case .datagram(var frame) = frame else { continue }
-                        frame.frame.finalize(success: false)
-                    default: continue
-                    }
-                }
+                self.cleanupReceivedFrames()
                 throw error
             }
             return packet
@@ -342,7 +357,6 @@ struct PacketParser: ~Copyable, PrefixedLoggable {
                 originalLength: originalLength
             )
         }
-        packet.framesReceived.reserveCapacity(1)
         return packet
     }
 

Sources/SwiftNetwork/QUIC/Preferences.swift

@@ -30,6 +30,7 @@ struct QUICPreferences {
     let initialMaxStreamBidirectionalLocalData: Int? = nil
     let maxConnectivityProbes: Int = 3
     let maxSentPacketsCapacity: Int = 512
+    let maxReceivedFramesCapacity: Int = 128
 
     private init() {}
 }
@@ -56,6 +57,7 @@ struct QUICPreferences: ~Copyable, Sendable {
     let quiclogDirectory: String
     let maxConnectivityProbes: Int
     let maxSentPacketsCapacity: Int
+    let maxReceivedFramesCapacity: Int
 
     // Flow Control
     let initialStreamReceiveSpace: Int?
@@ -116,6 +118,10 @@ struct QUICPreferences: ~Copyable, Sendable {
             "max_sent_packets_capacity",
             defaultValue: 512
         )
+        maxReceivedFramesCapacity = QUICPreferences.findSetting(
+            "max_received_frames_capacity",
+            defaultValue: 128
+        )
         quiclogDirectory = QUICPreferences.findSetting("quiclog_directory", defaultValue: "")
 
         // Flow Control

Sources/SwiftNetwork/QUIC/QUICConnection.swift

@@ -1720,7 +1720,7 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol,
 
         defer {
             // Make sure to always clean up any unprocessed frames when exiting
-            packet.cleanupReceivedFrames()
+            packetParser.cleanupReceivedFrames()
         }
 
         log(packet: &packet, coalesced: coalesced, outbound: false)
@@ -1768,7 +1768,7 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol,
         var isAckEliciting = false
         var isNonProbing = false
 
-        while let quicFrame = packet.framesReceived.popFirst() {
+        while let quicFrame = packetParser.framesReceived.popFirst() {
             if state == .initialReceived {
                 if !QUICFrame.isValidInInitial(frame: quicFrame) {
                     close(

Tests/QUICTests/QUICLayoutTests.swift

@@ -26,7 +26,7 @@ import XCTest
 final class QUICLayoutTests: XCTestCase {
 
     func testLayoutPacket() {
-        let packetSize = 188
+        let packetSize = 156
         let packetRecordSize = 201
         XCTAssertEqual(packetSize, MemoryLayout<Packet>.size)
         XCTAssertEqual(packetRecordSize, MemoryLayout<SentPacketRecord>.size)

Tests/QUICTests/QUICPacketTests.swift

@@ -55,23 +55,19 @@ final class PacketTests: XCTestCase {
     }
 
     func testCleanupReceivedFrames_drainsEntireDeque() {
-        var packet = Packet(
-            number: PacketNumber(1),
-            lastAcked: 0,
-            keyState: .phase0
-        )
-        packet.framesReceived.append(
+        var parser = PacketParser(logPrefixer: LogPrefixer("[PacketTests]"))
+        parser.framesReceived.append(
             .stream(frame: FrameStreamReceived(id: 0, offset: 0, data: [1, 2, 3]))
         )
-        packet.framesReceived.append(
+        parser.framesReceived.append(
             .stream(frame: FrameStreamReceived(id: 4, offset: 0, data: [4, 5, 6]))
         )
 
-        packet.cleanupReceivedFrames()
+        parser.cleanupReceivedFrames()
 
         // Drain leftover frames at the end of the test to avoid precondition failure in deinit.
         defer {
-            while let leftover = packet.framesReceived.popFirst() {
+            while let leftover = parser.framesReceived.popFirst() {
                 switch leftover {
                 case .crypto(var f): f.frame.finalize(success: false)
                 case .stream(var f): f.frame.finalize(success: false)
@@ -82,7 +78,7 @@ final class PacketTests: XCTestCase {
         }
 
         XCTAssertTrue(
-            packet.framesReceived.isEmpty,
+            parser.framesReceived.isEmpty,
             "cleanupReceivedFrames must drain all frames, not just the first"
         )
     }