integrate http2 state machine

Author: dnadoba

Diff for: Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1Connections.swift

@@ -618,9 +618,10 @@ extension HTTPConnectionPool {
             // TODO: improve algorithm to create connections uniformly across all preferred event loops
             // while paying attention to the number of queued request per event loop
             // Currently we start by creating new connections on the event loop with the most queued
-            // requests. If we have create a enough connections to cover all requests for the given
+            // requests. If we have created enough connections to cover all requests for the first
             // event loop we will continue with the event loop with the second most queued requests
-            // and so on and so forth. We do not need to sort the array because
+            // and so on and so forth. The `generalPurposeRequestCountGroupedByPreferredEventLoop`
+            // array is already ordered so we can just iterate over it without sorting by request count.
             let newGeneralPurposeConnections: [(Connection.ID, EventLoop)] = generalPurposeRequestCountGroupedByPreferredEventLoop
                 // we do not want to allocated intermediate arrays.
                 .lazy

Diff for: Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift

@@ -404,9 +404,9 @@ extension HTTPConnectionPool {
             self.connections.removeAll { connection in
                 switch connection.migrateToHTTP1(context: &context) {
                 case .removeConnection:
-                    return false
-                case .keepConnection:
                     return true
+                case .keepConnection:
+                    return false
                 }
             }
             return context
@@ -419,7 +419,7 @@ extension HTTPConnectionPool {
             self.connections.contains { $0.isActive }
         }
 
-        /// used in general purpose connection scenarios to check if at least one connection exist, or if should we create a new one
+        /// used in general purpose connection scenarios to check if at least one connection is starting or active for the given `eventLoop`
         var hasConnectionThatCanOrWillBeAbleToExecuteRequests: Bool {
             self.connections.contains { $0.canOrWillBeAbleToExecuteRequests }
         }
@@ -433,6 +433,32 @@ extension HTTPConnectionPool {
             }
         }
 
+        func hasActiveConnection(for eventLoop: EventLoop) -> Bool {
+            self.connections.contains {
+                $0.eventLoop === eventLoop && $0.isActive
+            }
+        }
+
+        /// used after backoff is done to determine if we need to create a new connection
+        func hasStartingOrActiveConnection() -> Bool {
+            self.connections.contains { connection in
+                connection.canOrWillBeAbleToExecuteRequests
+            }
+        }
+
+        /// used after backoff is done to determine if we need to create a new connection
+        /// - Parameters:
+        ///   - eventLoop: connection `EventLoop` to search for
+        /// - Returns: true if at least one connection is starting or active for the given `eventLoop`
+        func hasStartingOrActiveConnection(
+            for eventLoop: EventLoop
+        ) -> Bool {
+            self.connections.contains { connection in
+                connection.eventLoop === eventLoop &&
+                    connection.canOrWillBeAbleToExecuteRequests
+            }
+        }
+
         mutating func createNewConnection(on eventLoop: EventLoop) -> Connection.ID {
             assert(
                 !self.hasConnectionThatCanOrWillBeAbleToExecuteRequests(for: eventLoop),

Diff for: Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift

@@ -228,11 +228,22 @@ extension HTTPConnectionPool {
         private mutating func _newHTTP2ConnectionEstablished(_ connection: Connection, maxConcurrentStreams: Int) -> EstablishedAction {
             self.failedConsecutiveConnectionAttempts = 0
             self.lastConnectFailure = nil
-            let (index, context) = self.connections.newHTTP2ConnectionEstablished(
-                connection,
-                maxConcurrentStreams: maxConcurrentStreams
-            )
-            return self.nextActionForAvailableConnection(at: index, context: context)
+            if self.connections.hasActiveConnection(for: connection.eventLoop) {
+                guard let (index, _) = self.connections.failConnection(connection.id) else {
+                    preconditionFailure("we connection to a connection which we no nothing about")
+                }
+                self.connections.removeConnection(at: index)
+                return .init(
+                    request: .none,
+                    connection: .closeConnection(connection, isShutdown: .no)
+                )
+            } else {
+                let (index, context) = self.connections.newHTTP2ConnectionEstablished(
+                    connection,
+                    maxConcurrentStreams: maxConcurrentStreams
+                )
+                return self.nextActionForAvailableConnection(at: index, context: context)
+            }
         }
 
         private mutating func nextActionForAvailableConnection(
@@ -318,18 +329,35 @@ extension HTTPConnectionPool {
         private mutating func nextActionForFailedConnection(at index: Int, on eventLoop: EventLoop) -> Action {
             switch self.state {
             case .running:
-                let hasPendingRequest = !self.requests.isEmpty(for: eventLoop) || !self.requests.isEmpty(for: nil)
-                guard hasPendingRequest else {
+                // we do not know if we have created this connection for a request with a required
+                // event loop or not. However, we do not need this information and can infer
+                // if we need to create a new connection because we will only ever create one connection
+                // per event loop for required event loop requests and only need one connection for
+                // general purpose requests.
+
+                // we need to start a new on connection in two cases:
+                if
+                    // 1. if we have general purpose requests
+                    self.requests.isEmpty(for: nil) &&
+                    // and no connection starting or active
+                    self.connections.hasStartingOrActiveConnection() ||
+                    // 2. or if we have requests for a required event loop
+                    self.requests.isEmpty(for: eventLoop) &&
+                    // and no connection starting or active for the given event loop
+                    self.connections.hasStartingOrActiveConnection(for: eventLoop) {
+                    let (newConnectionID, previousEventLoop) = self.connections.createNewConnectionByReplacingClosedConnection(at: index)
+                    precondition(previousEventLoop === eventLoop)
+
+                    return .init(
+                        request: .none,
+                        connection: .createConnection(newConnectionID, on: eventLoop)
+                    )
+                } else {
+                    // otherwise we can remove the connection
+                    self.connections.removeConnection(at: index)
                     return .none
                 }
 
-                let (newConnectionID, previousEventLoop) = self.connections.createNewConnectionByReplacingClosedConnection(at: index)
-                precondition(previousEventLoop === eventLoop)
-
-                return .init(
-                    request: .none,
-                    connection: .createConnection(newConnectionID, on: eventLoop)
-                )
             case .shuttingDown(let unclean):
                 assert(self.requests.isEmpty)
                 self.connections.removeConnection(at: index)