Started work on h2

Author: fabianfett

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

@@ -12,7 +12,159 @@
 //
 //===----------------------------------------------------------------------===//
 
+import Logging
+import NIO
 import NIOConcurrencyHelpers
+import NIOHTTP1
+
+protocol HTTPConnectionPoolManagerDelegate: AnyObject {
+    func httpConnectionPoolManagerDidShutdown(_: HTTPConnectionPool.Manager, unclean: Bool)
+}
+
+extension HTTPConnectionPool {
+    final class Manager {
+        private typealias Key = ConnectionPool.Key
+
+        private var _pools: [Key: HTTPConnectionPool] = [:]
+        private let lock = Lock()
+
+        private let sslContextCache = SSLContextCache()
+
+        enum State {
+            case active
+            case shuttingDown(unclean: Bool)
+            case shutDown
+        }
+
+        let eventLoopGroup: EventLoopGroup
+        let configuration: HTTPClient.Configuration
+        let connectionIDGenerator = Connection.ID.globalGenerator
+        let logger: Logger
+
+        /// A delegate to inform about the pools managers shutdown
+        ///
+        /// NOTE: Normally we create retain cycles in SwiftNIO code that we break on shutdown. However we wan't to inform
+        ///      users that they must call `shutdown` on their AsyncHTTPClient. The best way to make them aware is with
+        ///      a `preconditionFailure` in the HTTPClient's `deinit`. If we create a retain cycle here, the
+        ///      `HTTPClient`'s `deinit` can never be reached. Instead the `HTTPClient` would leak.
+        ///
+        ///      The delegate is not thread safe at all. This only works if the HTTPClient sets itself as a delegate in its own
+        ///      init.
+        weak var delegate: HTTPConnectionPoolManagerDelegate?
+
+        private var state: State = .active
+
+        init(eventLoopGroup: EventLoopGroup,
+             configuration: HTTPClient.Configuration,
+             backgroundActivityLogger logger: Logger) {
+            self.eventLoopGroup = eventLoopGroup
+            self.configuration = configuration
+            self.logger = logger
+        }
+
+        deinit {
+            guard case .shutDown = self.state else {
+                preconditionFailure("Manager must be shutdown before deinit")
+            }
+        }
+
+        func execute(request: HTTPRequestTask) {
+            let key = Key(request.request)
+
+            let poolResult = self.lock.withLock { () -> Result<HTTPConnectionPool, HTTPClientError> in
+                guard case .active = self.state else {
+                    return .failure(HTTPClientError.alreadyShutdown)
+                }
+
+                if let pool = self._pools[key] {
+                    return .success(pool)
+                }
+
+                let pool = HTTPConnectionPool(
+                    eventLoopGroup: self.eventLoopGroup,
+                    sslContextCache: self.sslContextCache,
+                    tlsConfiguration: request.request.tlsConfiguration,
+                    clientConfiguration: self.configuration,
+                    key: key,
+                    delegate: self,
+                    idGenerator: self.connectionIDGenerator,
+                    logger: self.logger
+                )
+                self._pools[key] = pool
+                return .success(pool)
+            }
+
+            switch poolResult {
+            case .success(let pool):
+                pool.execute(request: request)
+            case .failure(let error):
+                request.fail(error)
+            }
+        }
+
+        func shutdown() {
+            let pools = self.lock.withLock { () -> [Key: HTTPConnectionPool] in
+                guard case .active = self.state else {
+                    preconditionFailure("PoolManager already shutdown")
+                }
+
+                // If there aren't any pools, we can mark the pool as shut down right away.
+                if self._pools.isEmpty {
+                    self.state = .shutDown
+                } else {
+                    self.state = .shuttingDown(unclean: false)
+                }
+
+                return self._pools
+            }
+
+            // if no pools are returned, the manager is already shutdown completely. Inform the
+            // delegate. This is a very clean shutdown...
+            if pools.isEmpty {
+                self.delegate?.httpConnectionPoolManagerDidShutdown(self, unclean: false)
+                return
+            }
+
+            pools.values.forEach { pool in
+                pool.shutdown()
+            }
+        }
+    }
+}
+
+extension HTTPConnectionPool.Manager: HTTPConnectionPoolDelegate {
+    enum CloseAction {
+        case close(unclean: Bool)
+        case wait
+    }
+
+    func connectionPoolDidShutdown(_ pool: HTTPConnectionPool, unclean: Bool) {
+        let closeAction = self.lock.withLock { () -> CloseAction in
+            guard case .shuttingDown(let soFarUnclean) = self.state else {
+                preconditionFailure("Why are pools shutting down, if the manager did not give a signal")
+            }
+
+            guard self._pools.removeValue(forKey: pool.key) === pool else {
+                preconditionFailure("Expected that the pool was ")
+            }
+
+            if self._pools.isEmpty {
+                self.state = .shutDown
+                return .close(unclean: soFarUnclean || unclean)
+            } else {
+                self.state = .shuttingDown(unclean: soFarUnclean || unclean)
+                return .wait
+            }
+        }
+
+        switch closeAction {
+        case .close(unclean: let unclean):
+            self.delegate?.httpConnectionPoolManagerDidShutdown(self, unclean: unclean)
+        case .wait:
+            break
+        }
+    }
+}
 
 extension HTTPConnectionPool.Connection.ID {
     static var globalGenerator = Generator()

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)
 }