[HTTP2] Integrate HTTP2StateMachine into HTTPConnectionPool.StateMachine (#462)

Author: dnadoba

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

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

@@ -45,18 +45,6 @@ extension HTTPConnectionPool {
             self.requests = RequestQueue()
         }
 
-        mutating func migrateFromHTTP2(
-            http2State: HTTP2StateMachine,
-            newHTTP1Connection: Connection
-        ) -> Action {
-            self.migrateFromHTTP2(
-                http1Connections: http2State.http1Connections,
-                http2Connections: http2State.connections,
-                requests: http2State.requests,
-                newHTTP1Connection: newHTTP1Connection
-            )
-        }
-
         mutating func migrateFromHTTP2(
             http1Connections: HTTP1Connections? = nil,
             http2Connections: HTTP2Connections,

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,20 +419,46 @@ 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, backing off or active
         var hasConnectionThatCanOrWillBeAbleToExecuteRequests: Bool {
             self.connections.contains { $0.canOrWillBeAbleToExecuteRequests }
         }
 
         /// used in eventLoop scenarios. does at least one connection exist for this eventLoop, or should we create a new one?
         /// - Parameter eventLoop: connection `EventLoop` to search for
-        /// - Returns: true if at least one connection is starting or active for the given `eventLoop`
+        /// - Returns: true if at least one connection is starting, backing off or active for the given `eventLoop`
         func hasConnectionThatCanOrWillBeAbleToExecuteRequests(for eventLoop: EventLoop) -> Bool {
             self.connections.contains {
                 $0.eventLoop === eventLoop && $0.canOrWillBeAbleToExecuteRequests
             }
         }
 
+        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),

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

@@ -49,20 +49,6 @@ extension HTTPConnectionPool {
             self.connections = HTTP2Connections(generator: idGenerator)
         }
 
-        mutating func migrateFromHTTP1(
-            http1State: HTTP1StateMachine,
-            newHTTP2Connection: Connection,
-            maxConcurrentStreams: Int
-        ) -> Action {
-            self.migrateFromHTTP1(
-                http1Connections: http1State.connections,
-                http2Connections: http1State.http2Connections,
-                requests: http1State.requests,
-                newHTTP2Connection: newHTTP2Connection,
-                maxConcurrentStreams: maxConcurrentStreams
-            )
-        }
-
         mutating func migrateFromHTTP1(
             http1Connections: HTTP1Connections,
             http2Connections: HTTP2Connections? = nil,
@@ -228,11 +214,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 have established a new connection that we know 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,8 +315,28 @@ 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:
+                let needGeneralPurposeConnection =
+                    // 1. if we have general purpose requests
+                    !self.requests.isEmpty(for: nil) &&
+                    // and no connection starting or active
+                    !self.connections.hasStartingOrActiveConnection()
+
+                let needRequiredEventLoopConnection =
+                    // 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)
+
+                guard needGeneralPurposeConnection || needRequiredEventLoopConnection else {
+                    // otherwise we can remove the connection
+                    self.connections.removeConnection(at: index)
                     return .none
                 }
 
@@ -330,6 +347,7 @@ extension HTTPConnectionPool {
                     request: .none,
                     connection: .createConnection(newConnectionID, on: eventLoop)
                 )
+
             case .shuttingDown(let unclean):
                 assert(self.requests.isEmpty)
                 self.connections.removeConnection(at: index)