Switch to new HTTPConnectionPool

Author: fabianfett

Diff for: Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift

@@ -29,7 +29,6 @@ protocol HTTPConnectionPoolDelegate {
 }
 
 class HTTPConnectionPool {
-    
     struct Connection: Equatable {
         typealias ID = Int
 

Diff for: Sources/AsyncHTTPClient/HTTPClient.swift

@@ -66,7 +66,7 @@ public class HTTPClient {
     public let eventLoopGroup: EventLoopGroup
     let eventLoopGroupProvider: EventLoopGroupProvider
     let configuration: Configuration
-    let pool: ConnectionPool
+    let poolManager: HTTPConnectionPool.Manager
     var state: State
     private let stateLock = Lock()
 
@@ -108,14 +108,20 @@ public class HTTPClient {
             #endif
         }
         self.configuration = configuration
-        self.pool = ConnectionPool(configuration: configuration,
-                                   backgroundActivityLogger: backgroundActivityLogger)
+        self.poolManager = HTTPConnectionPool.Manager(
+            eventLoopGroup: self.eventLoopGroup,
+            configuration: self.configuration,
+            backgroundActivityLogger: backgroundActivityLogger
+        )
         self.state = .upAndRunning
+
+        self.poolManager.delegate = self
     }
 
     deinit {
-        assert(self.pool.count == 0)
-        assert(self.state == .shutDown, "Client not shut down before the deinit. Please call client.syncShutdown() when no longer needed.")
+        guard case .shutDown = self.state else {
+            preconditionFailure("Client not shut down before the deinit. Please call client.syncShutdown() when no longer needed.")
+        }
     }
 
     /// Shuts down the client and `EventLoopGroup` if it was created by the client.
@@ -189,36 +195,17 @@ public class HTTPClient {
     private func shutdown(requiresCleanClose: Bool, queue: DispatchQueue, _ callback: @escaping (Error?) -> Void) {
         do {
             try self.stateLock.withLock {
-                if self.state != .upAndRunning {
+                guard case .upAndRunning = self.state else {
                     throw HTTPClientError.alreadyShutdown
                 }
-                self.state = .shuttingDown
+                self.state = .shuttingDown(requiresCleanClose: requiresCleanClose, callback: callback)
             }
         } catch {
             callback(error)
             return
         }
 
-        self.pool.close(on: self.eventLoopGroup.next()).whenComplete { result in
-            var closeError: Error?
-            switch result {
-            case .failure(let error):
-                closeError = error
-            case .success(let cleanShutdown):
-                if !cleanShutdown, requiresCleanClose {
-                    closeError = HTTPClientError.uncleanShutdown
-                }
-
-                self.shutdownEventLoop(queue: queue) { eventLoopError in
-                    // we prioritise .uncleanShutdown here
-                    if let error = closeError {
-                        callback(error)
-                    } else {
-                        callback(eventLoopError)
-                    }
-                }
-            }
-        }
+        self.poolManager.shutdown()
     }
 
     /// Execute `GET` request using specified URL.
@@ -490,7 +477,7 @@ public class HTTPClient {
         let taskEL: EventLoop
         switch eventLoopPreference.preference {
         case .indifferent:
-            taskEL = self.pool.associatedEventLoop(for: ConnectionPool.Key(request)) ?? self.eventLoopGroup.next()
+            taskEL = self.eventLoopGroup.next()
         case .delegate(on: let eventLoop):
             precondition(self.eventLoopGroup.makeIterator().contains { $0 === eventLoop }, "Provided EventLoop must be part of clients EventLoopGroup.")
             taskEL = eventLoop
@@ -538,77 +525,30 @@ public class HTTPClient {
         }
 
         let task = Task<Delegate.Response>(eventLoop: taskEL, logger: logger)
-        let setupComplete = taskEL.makePromise(of: Void.self)
-        let connection = self.pool.getConnection(request,
-                                                 preference: eventLoopPreference,
-                                                 taskEventLoop: taskEL,
-                                                 deadline: deadline,
-                                                 setupComplete: setupComplete.futureResult,
-                                                 logger: logger)
-
-        let taskHandler = TaskHandler(task: task,
-                                      kind: request.kind,
-                                      delegate: delegate,
-                                      redirectHandler: redirectHandler,
-                                      ignoreUncleanSSLShutdown: self.configuration.ignoreUncleanSSLShutdown,
-                                      logger: logger)
-
-        connection.flatMap { connection -> EventLoopFuture<Void> in
-            logger.debug("got connection for request",
-                         metadata: ["ahc-connection": "\(connection)",
-                                    "ahc-request": "\(request.method) \(request.url)",
-                                    "ahc-channel-el": "\(connection.channel.eventLoop)",
-                                    "ahc-task-el": "\(taskEL)"])
-
-            let channel = connection.channel
-
-            func prepareChannelForTask0() -> EventLoopFuture<Void> {
-                do {
-                    let syncPipelineOperations = channel.pipeline.syncOperations
-
-                    if let timeout = self.resolve(timeout: self.configuration.timeout.read, deadline: deadline) {
-                        try syncPipelineOperations.addHandler(IdleStateHandler(readTimeout: timeout))
-                    }
-
-                    try syncPipelineOperations.addHandler(taskHandler)
-                } catch {
-                    connection.release(closing: true, logger: logger)
-                    return channel.eventLoop.makeFailedFuture(error)
-                }
-
-                task.setConnection(connection)
 
-                let isCancelled = task.lock.withLock {
-                    task.cancelled
-                }
-
-                if !isCancelled {
-                    return channel.writeAndFlush(request).flatMapError { _ in
-                        // At this point the `TaskHandler` will already be present
-                        // to handle the failure and pass it to the `promise`
-                        channel.eventLoop.makeSucceededVoidFuture()
-                    }
-                } else {
-                    return channel.eventLoop.makeSucceededVoidFuture()
-                }
+        let requestBag = RequestBag(
+            request: request,
+            eventLoopPreference: eventLoopPreference,
+            task: task,
+            redirectHandler: redirectHandler,
+            connectionDeadline: .now() + (self.configuration.timeout.connect ?? .seconds(10)),
+            idleReadTimeout: self.configuration.timeout.read,
+            delegate: delegate
+        )
+
+        var deadlineSchedule: Scheduled<Void>?
+        if let deadline = deadline {
+            deadlineSchedule = taskEL.scheduleTask(deadline: deadline) {
+                requestBag.fail(HTTPClientError.deadlineExceeded)
             }
 
-            if channel.eventLoop.inEventLoop {
-                return prepareChannelForTask0()
-            } else {
-                return channel.eventLoop.flatSubmit {
-                    return prepareChannelForTask0()
-                }
-            }
-        }.always { _ in
-            setupComplete.succeed(())
-        }.whenFailure { error in
-            taskHandler.callOutToDelegateFireAndForget { task in
-                delegate.didReceiveError(task: task, error)
+            task.promise.futureResult.whenComplete { _ in
+                deadlineSchedule?.cancel()
             }
-            task.promise.fail(error)
         }
 
+        self.poolManager.execute(request: requestBag)
+
         return task
     }
 
@@ -815,7 +755,7 @@ public class HTTPClient {
 
     enum State {
         case upAndRunning
-        case shuttingDown
+        case shuttingDown(requiresCleanClose: Bool, callback: (Error?) -> Void)
         case shutDown
     }
 }
@@ -882,6 +822,22 @@ extension HTTPClient.Configuration {
     }
 }
 
+extension HTTPClient: HTTPConnectionPoolManagerDelegate {
+    func httpConnectionPoolManagerDidShutdown(_: HTTPConnectionPool.Manager, unclean: Bool) {
+        let (callback, error) = self.stateLock.withLock { () -> ((Error?) -> Void, Error?) in
+            guard case .shuttingDown(let requiresClean, callback: let callback) = self.state else {
+                preconditionFailure("Why did the pool manager shut down, if it was not instructed to")
+            }
+
+            self.state = .shutDown
+            let error: Error? = (requiresClean && unclean) ? HTTPClientError.uncleanShutdown : nil
+            return (callback, error)
+        }
+
+        callback(error)
+    }
+}
+
 /// Possible client errors.
 public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
     private enum Code: Equatable {
@@ -909,6 +865,7 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
         case incompatibleHeaders
         case connectTimeout
         case getConnectionFromPoolTimeout
+        case deadlineExceeded
     }
 
     private var code: Code
@@ -973,4 +930,6 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
     ///  - A connection could not be created within the timout period.
     ///  - Tasks are not processed fast enough on the existing connections, to process all waiters in time
     public static let getConnectionFromPoolTimeout = HTTPClientError(code: .getConnectionFromPoolTimeout)
+    /// The request deadline was exceeded. The request was cancelled because of this.
+    public static let deadlineExceeded = HTTPClientError(code: .deadlineExceeded)
 }

Diff for: Sources/AsyncHTTPClient/HTTPHandler.swift

@@ -636,18 +636,33 @@ extension HTTPClient {
         public let eventLoop: EventLoop
 
         let promise: EventLoopPromise<Response>
-        var completion: EventLoopFuture<Void>
-        var connection: Connection?
-        var cancelled: Bool
-        let lock: Lock
+
+        var connection: HTTPConnectionPool.Connection? {
+            self.lock.withLock { self._connection }
+        }
+
+        var isCancelled: Bool {
+            self.lock.withLock { self._isCancelled }
+        }
+
+        var taskDelegate: HTTPClientTaskDelegate? {
+            get {
+                self.lock.withLock { self._taskDelegate }
+            }
+            set {
+                self.lock.withLock { self._taskDelegate = newValue }
+            }
+        }
+
+        private var _connection: HTTPConnectionPool.Connection?
+        private var _isCancelled: Bool = false
+        private var _taskDelegate: HTTPClientTaskDelegate?
+        private let lock = Lock()
         let logger: Logger // We are okay to store the logger here because a Task is for only one request.
 
         init(eventLoop: EventLoop, logger: Logger) {
             self.eventLoop = eventLoop
             self.promise = eventLoop.makePromise()
-            self.completion = self.promise.futureResult.map { _ in }
-            self.cancelled = false
-            self.lock = Lock()
             self.logger = logger
         }
 
@@ -672,69 +687,30 @@ extension HTTPClient {
 
         /// Cancels the request execution.
         public func cancel() {
-            let channel: Channel? = self.lock.withLock {
-                if !self.cancelled {
-                    self.cancelled = true
-                    return self.connection?.channel
-                } else {
-                    return nil
-                }
+            let taskDelegate = self.lock.withLock { () -> HTTPClientTaskDelegate? in
+                self._isCancelled = true
+                return self._taskDelegate
             }
-            channel?.triggerUserOutboundEvent(TaskCancelEvent(), promise: nil)
+
+            taskDelegate?.cancel()
         }
 
-        @discardableResult
-        func setConnection(_ connection: Connection) -> Connection {
+        func setConnection(_ connection: HTTPConnectionPool.Connection) {
             return self.lock.withLock {
-                self.connection = connection
-                if self.cancelled {
-                    connection.channel.triggerUserOutboundEvent(TaskCancelEvent(), promise: nil)
-                }
-                return connection
+                self._connection = connection
             }
         }
 
         func succeed<Delegate: HTTPClientResponseDelegate>(promise: EventLoopPromise<Response>?,
                                                            with value: Response,
                                                            delegateType: Delegate.Type,
                                                            closing: Bool) {
-            self.releaseAssociatedConnection(delegateType: delegateType,
-                                             closing: closing).whenSuccess {
-                promise?.succeed(value)
-            }
+            promise?.succeed(value)
         }
 
         func fail<Delegate: HTTPClientResponseDelegate>(with error: Error,
                                                         delegateType: Delegate.Type) {
-            if let connection = self.connection {
-                self.releaseAssociatedConnection(delegateType: delegateType, closing: true)
-                    .whenSuccess {
-                        self.promise.fail(error)
-                        connection.channel.close(promise: nil)
-                    }
-            } else {
-                // this is used in tests where we don't want to bootstrap the whole connection pool
-                self.promise.fail(error)
-            }
-        }
-
-        func releaseAssociatedConnection<Delegate: HTTPClientResponseDelegate>(delegateType: Delegate.Type,
-                                                                               closing: Bool) -> EventLoopFuture<Void> {
-            if let connection = self.connection {
-                // remove read timeout handler
-                return connection.removeHandler(IdleStateHandler.self).flatMap {
-                    connection.removeHandler(TaskHandler<Delegate>.self)
-                }.map {
-                    connection.release(closing: closing, logger: self.logger)
-                }.flatMapError { error in
-                    fatalError("Couldn't remove taskHandler: \(error)")
-                }
-            } else {
-                // TODO: This seems only reached in some internal unit test
-                // Maybe there could be a better handling in the future to make
-                // it an error outside of testing contexts
-                return self.eventLoop.makeSucceededFuture(())
-            }
+            self.promise.fail(error)
         }
     }
 }
@@ -1071,9 +1047,7 @@ extension TaskHandler: ChannelDuplexHandler {
                 break
             case .redirected(let head, let redirectURL):
                 self.state = .endOrError
-                self.task.releaseAssociatedConnection(delegateType: Delegate.self, closing: self.closing).whenSuccess {
-                    self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise: self.task.promise)
-                }
+                self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise: self.task.promise)
             default:
                 self.state = .bufferedEnd
                 self.handleReadForDelegate(response, context: context)

Diff for: Sources/AsyncHTTPClient/RequestBag.swift

@@ -86,14 +86,14 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate>: HTTPRequestTask {
         self.idleReadTimeout = idleReadTimeout
         self.delegate = delegate
 
-//        self.task.taskDelegate = self
-//        self.task.futureResult.whenComplete { _ in
-//            self.task.taskDelegate = nil
-//        }
+        self.task.taskDelegate = self
+        self.task.futureResult.whenComplete { _ in
+            self.task.taskDelegate = nil
+        }
     }
 
     func willBeExecutedOnConnection(_ connection: HTTPConnectionPool.Connection) {
-//        self.task.setConnection(connection)
+        self.task.setConnection(connection)
     }
 
     func requestWasQueued(_ queuer: HTTP1RequestQueuer) {