fix review comments

- new connections that are created after the http2 -> http1 migration should now more closely mimic the behaviour of the http1 state machine
- API bounderies are now clearer

Author: dnadoba

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1Connections.swift

@@ -278,10 +278,6 @@ extension HTTPConnectionPool {
             self.overflowIndex < self.maximumConcurrentConnections
         }
 
-        private var maximumAdditionalGeneralPurposeConnections: Int {
-            self.maximumConcurrentConnections - (self.overflowIndex - 1)
-        }
-
         var startingGeneralPurposeConnections: Int {
             var connecting = 0
             for connectionState in self.connections[0..<self.overflowIndex] {
@@ -545,66 +541,84 @@ extension HTTPConnectionPool {
 
         /// We only handle starting and backing off connection here.
         /// All already running connections must be handled by the enclosing state machine.
-        /// We will also create new connections for `requiredEventLoopsOfPendingRequests`
-        /// if we do not already have a connection that can or will be able to execute requests on the given event loop.
-        /// In addition, we also create more general purpose connections if we do not have enough to execute
-        /// all requests on the given `preferredEventLoopsOfPendingGeneralPurposeRequests`
-        /// until we reach `maximumConcurrentConnections`.
         /// - Parameters:
         ///   - starting: starting HTTP connections from previous state machine
         ///   - backingOff: backing off HTTP connections from previous state machine
-        ///   - requiredEventLoopsForPendingRequests: event loops for which we have requests with a required event loop. Duplicates are not allowed.
-        ///   - preferredEventLoopsOfPendingGeneralPurposeRequests: the preferred event loop of all pending requests without a required event loop
-        ///   - Returns: new connections that need to be created
-        mutating func migrateFromHTTP2<PreferredEventLoopSequence>(
+        mutating func migrateFromHTTP2(
             starting: [(Connection.ID, EventLoop)],
-            backingOff: [(Connection.ID, EventLoop)],
-            requiredEventLoopsOfPendingRequests: [EventLoop],
-            preferredEventLoopsOfPendingGeneralPurposeRequests: PreferredEventLoopSequence
-        ) -> [(Connection.ID, EventLoop)] where PreferredEventLoopSequence: Sequence, PreferredEventLoopSequence.Element == EventLoop {
+            backingOff: [(Connection.ID, EventLoop)]
+        ) {
             for (connectionID, eventLoop) in starting {
                 let newConnection = HTTP1ConnectionState(connectionID: connectionID, eventLoop: eventLoop)
                 self.connections.insert(newConnection, at: self.overflowIndex)
-                self.overflowIndex = self.connections.index(after: self.overflowIndex)
+                /// If we can grow, we mark the connection as a general purpose connection.
+                /// Otherwise, it will be an overflow connection which is only used once for requests with a required event loop
+                if self.canGrow {
+                    self.overflowIndex = self.connections.index(after: self.overflowIndex)
+                }
             }
 
             for (connectionID, eventLoop) in backingOff {
                 var backingOffConnection = HTTP1ConnectionState(connectionID: connectionID, eventLoop: eventLoop)
                 // TODO: Maybe we want to add a static init for backing off connections to HTTP1ConnectionState
                 backingOffConnection.failedToConnect()
                 self.connections.insert(backingOffConnection, at: self.overflowIndex)
-                self.overflowIndex = self.connections.index(after: self.overflowIndex)
+                /// If we can grow, we mark the connection as a general purpose connection.
+                /// Otherwise, it will be an overflow connection which is only used once for requests with a required event loop
+                if self.canGrow {
+                    self.overflowIndex = self.connections.index(after: self.overflowIndex)
+                }
             }
+        }
 
-            // create new connections for requests with a required event loop
-            let eventLoopsWithConnectionThatCanOrWillBeAbleToExecuteRequests = Set(
-                self.connections.lazy
-                    .filter {
-                        $0.canOrWillBeAbleToExecuteRequests
-                    }.map {
-                        $0.eventLoop.id
-                    }
+        /// We will create new connections for each `requiredEventLoopOfPendingRequests`
+        /// In addition, we also create more general purpose connections if we do not have enough to execute
+        /// all requests on the given `preferredEventLoopsOfPendingGeneralPurposeRequests`
+        /// until we reach `maximumConcurrentConnections
+        /// - Parameters:
+        ///   - requiredEventLoopsForPendingRequests: event loops for which we have requests with a required event loop. Duplicates are not allowed.
+        ///   - generalPurposeRequestCountPerPreferredEventLoop: request count with no required event loop, grouped by preferred event loop and ordered descending by number of requests
+        /// - Returns: new connections that need to be created
+        mutating func createConnectionsAfterMigrationIfNeeded(
+            requiredEventLoopOfPendingRequests: [(EventLoop, Int)],
+            generalPurposeRequestCountGroupedByPreferredEventLoop: [(EventLoop, Int)]
+        ) -> [(Connection.ID, EventLoop)] {
+            /// create new connections for requests with a required event loop
+
+            /// we may already start connections for those requests and do not want to start to many
+            let startingRequiredEventLoopConnectionCount = Dictionary(
+                self.connections[self.overflowIndex..<self.connections.endIndex].lazy.map {
+                    ($0.eventLoop.id, 1)
+                },
+                uniquingKeysWith: +
             )
-            var createConnections = requiredEventLoopsOfPendingRequests.compactMap { eventLoop -> (Connection.ID, EventLoop)? in
-                guard !eventLoopsWithConnectionThatCanOrWillBeAbleToExecuteRequests.contains(eventLoop.id)
-                else { return nil }
-                let connectionID = self.createNewOverflowConnection(on: eventLoop)
-                return (connectionID, eventLoop)
-            }
-
-            // create new connections for requests without a required event loop
-            createConnections.append(
-                contentsOf: preferredEventLoopsOfPendingGeneralPurposeRequests.lazy
-                    /// we do not want to start additional connection for requests for which we already have starting or backing off connections
-                    .dropFirst(self.startingGeneralPurposeConnections)
-                    /// if we have additional requests, we will start more connections while respecting the user defined `maximumConcurrentConnections`
-                    .prefix(self.maximumAdditionalGeneralPurposeConnections)
-                    .map { eventLoop in
-                        (self.createNewConnection(on: eventLoop), eventLoop)
+
+            var connectionToCreate = requiredEventLoopOfPendingRequests
+                .flatMap { (eventLoop, requestCount) -> [(Connection.ID, EventLoop)] in
+                    let connectionsToStart = max(requestCount - startingRequiredEventLoopConnectionCount[eventLoop.id, default: 0], 0)
+                    return (0..<connectionsToStart).lazy.map { _ in
+                        (self.createNewOverflowConnection(on: eventLoop), eventLoop)
                     }
-            )
+                }
+
+            /// create new connections for requests without a required event loop
+
+            /// we may already start connections for those requests and do not want to start to many
+            /// this integer keeps track of the number of connections which do not yet have any requests assigned
+            var unusedGeneralPurposeConnections = self.startingGeneralPurposeConnections
+
+            outerLoop: for (eventLoop, requestCount) in generalPurposeRequestCountGroupedByPreferredEventLoop {
+                let connectionsToStart = max(requestCount - unusedGeneralPurposeConnections, 0)
+                unusedGeneralPurposeConnections -= min(requestCount, unusedGeneralPurposeConnections)
+                for _ in 0..<connectionsToStart {
+                    guard self.canGrow else {
+                        break outerLoop
+                    }
+                    connectionToCreate.append((self.createNewConnection(on: eventLoop), eventLoop))
+                }
+            }
 
-            return createConnections
+            return connectionToCreate
         }
 
         // MARK: Shutdown

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift

@@ -96,13 +96,15 @@ extension HTTPConnectionPool {
 
             var http2Connections = http2Connections
             let migration = http2Connections.migrateToHTTP1()
-            let createConnections = self.connections.migrateFromHTTP2(
+
+            self.connections.migrateFromHTTP2(
                 starting: migration.starting,
-                backingOff: migration.backingOff,
-                requiredEventLoopsOfPendingRequests: requests.eventLoopsWithPendingRequests(),
-                preferredEventLoopsOfPendingGeneralPurposeRequests: requests.generalPurposeQueue.lazy.map {
-                    $0.preferredEventLoop
-                }
+                backingOff: migration.backingOff
+            )
+
+            let createConnections = self.connections.createConnectionsAfterMigrationIfNeeded(
+                requiredEventLoopOfPendingRequests: requests.requestCountGroupedByRequiredEventLoop(),
+                generalPurposeRequestCountGroupedByPreferredEventLoop: requests.generalPurposeRequestCountGroupedByPreferredEventLoop()
             )
 
             if !http2Connections.isEmpty {

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift

@@ -348,18 +348,13 @@ extension HTTPConnectionPool {
 
         /// We only handle starting and backing off connection here.
         /// All already running connections must be handled by the enclosing state machine.
-        /// We will also create new connections for `requiredEventLoopsOfPendingRequests`
-        /// if we do not already have a connection that can or will be able to execute requests on the given event loop.
         /// - Parameters:
         ///   - starting: starting HTTP connections from previous state machine
         ///   - backingOff: backing off HTTP connections from previous state machine
-        ///   - requiredEventLoopsForPendingRequests: event loops for which we have requests with a required event loop. Duplicates are not allowed.
-        /// - Returns: new connections that need to be created
         mutating func migrateFromHTTP1(
             starting: [(Connection.ID, EventLoop)],
-            backingOff: [(Connection.ID, EventLoop)],
-            requiredEventLoopsOfPendingRequests: [EventLoop]
-        ) -> [(Connection.ID, EventLoop)] {
+            backingOff: [(Connection.ID, EventLoop)]
+        ) {
             for (connectionID, eventLoop) in starting {
                 let newConnection = HTTP2ConnectionState(connectionID: connectionID, eventLoop: eventLoop)
                 self.connections.append(newConnection)
@@ -371,7 +366,16 @@ extension HTTPConnectionPool {
                 backingOffConnection.failedToConnect()
                 self.connections.append(backingOffConnection)
             }
+        }
 
+        /// We will create new connections for `requiredEventLoopsOfPendingRequests`
+        /// if we do not already have a connection that can or will be able to execute requests on the given event loop.
+        /// - Parameters:
+        ///   - requiredEventLoopsForPendingRequests: event loops for which we have requests with a required event loop. Duplicates are not allowed.
+        /// - Returns: new connections that need to be created
+        mutating func createConnectionsAfterMigrationIfNeeded(
+            requiredEventLoopsOfPendingRequests: [EventLoop]
+        ) -> [(Connection.ID, EventLoop)] {
             // create new connections for requests with a required event loop
             let eventLoopsWithConnectionThatCanOrWillBeAbleToExecuteRequests = Set(
                 self.connections.lazy

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift

@@ -104,9 +104,12 @@ extension HTTPConnectionPool {
 
             var http1Connections = http1Connections // make http1Connections mutable
             let context = http1Connections.migrateToHTTP2()
-            let createConnections = self.connections.migrateFromHTTP1(
+            self.connections.migrateFromHTTP1(
                 starting: context.starting,
-                backingOff: context.backingOff,
+                backingOff: context.backingOff
+            )
+
+            let createConnections = self.connections.createConnectionsAfterMigrationIfNeeded(
                 requiredEventLoopsOfPendingRequests: requests.eventLoopsWithPendingRequests()
             )
 

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+RequestQueue.swift

@@ -14,10 +14,25 @@
 
 import NIOCore
 
+private struct HashableEventLoop: Hashable {
+    static func == (lhs: HashableEventLoop, rhs: HashableEventLoop) -> Bool {
+        lhs.eventLoop === rhs.eventLoop
+    }
+
+    init(_ eventLoop: EventLoop) {
+        self.eventLoop = eventLoop
+    }
+
+    let eventLoop: EventLoop
+    func hash(into hasher: inout Hasher) {
+        self.eventLoop.id.hash(into: &hasher)
+    }
+}
+
 extension HTTPConnectionPool {
     /// A struct to store all queued requests.
     struct RequestQueue {
-        private(set) var generalPurposeQueue: CircularBuffer<Request>
+        private var generalPurposeQueue: CircularBuffer<Request>
         private var eventLoopQueues: [EventLoopID: CircularBuffer<Request>]
 
         init() {
@@ -132,8 +147,40 @@ extension HTTPConnectionPool {
             return nil
         }
 
+        /// - Returns: event loops with at least one request with a required event loop
         func eventLoopsWithPendingRequests() -> [EventLoop] {
-            self.eventLoopQueues.compactMap { $0.value.first?.requiredEventLoop }
+            self.eventLoopQueues.compactMap {
+                /// all requests in `eventLoopQueues` are guaranteed to have a `requiredEventLoop`
+                /// however, a queue can be empty
+                $0.value.first?.requiredEventLoop!
+            }
+        }
+
+        /// - Returns: request count for requests with required event loop, grouped by required event loop without any particular order
+        func requestCountGroupedByRequiredEventLoop() -> [(EventLoop, Int)] {
+            self.eventLoopQueues.values.compactMap { requests -> (EventLoop, Int)? in
+                /// all requests in `eventLoopQueues` are guaranteed to have a `requiredEventLoop`,
+                /// however, a queue can be empty
+                guard let requiredEventLoop = requests.first?.requiredEventLoop! else {
+                    return nil
+                }
+                return (requiredEventLoop, requests.count)
+            }
+        }
+
+        /// - Returns: request count with **no** required event loop, grouped by preferred event loop and ordered descending by number of requests
+        func generalPurposeRequestCountGroupedByPreferredEventLoop() -> [(EventLoop, Int)] {
+            let requestCountPerEventLoop = Dictionary(
+                self.generalPurposeQueue.lazy.map { request in
+                    (HashableEventLoop(request.preferredEventLoop), 1)
+                },
+                uniquingKeysWith: +
+            )
+            return requestCountPerEventLoop.lazy
+                .map { ($0.key.eventLoop, $0.value) }
+                .sorted { lhs, rhs in
+                    lhs.1 > rhs.1
+                }
         }
     }
 }

Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1ConnectionsTest+XCTest.swift

@@ -39,6 +39,8 @@ extension HTTPConnectionPool_HTTP1ConnectionsTests {
             ("testMigrationFromHTTP2WithPendingRequestsWithRequiredEventLoop", testMigrationFromHTTP2WithPendingRequestsWithRequiredEventLoop),
             ("testMigrationFromHTTP2WithPendingRequestsWithPreferredEventLoop", testMigrationFromHTTP2WithPendingRequestsWithPreferredEventLoop),
             ("testMigrationFromHTTP2WithAlreadyLeasedHTTP1Connection", testMigrationFromHTTP2WithAlreadyLeasedHTTP1Connection),
+            ("testMigrationFromHTTP2WithMoreStartingConnectionsThanMaximumAllowedConccurentConnections", testMigrationFromHTTP2WithMoreStartingConnectionsThanMaximumAllowedConccurentConnections),
+            ("testMigrationFromHTTP2StartsEnoghOverflowConnectionsForRequiredEventLoopRequests", testMigrationFromHTTP2StartsEnoghOverflowConnectionsForRequiredEventLoopRequests),
         ]
     }
 }