Fixes #234 by removing setter on internal ConnectionsState so modification (#311)

ya-pulser · web-flow · commit e401a2801cf4 · 2020-10-02T14:31:25.000+01:00
allowed only using exposed API.

Motivation:

Having a setter for internal state of ConnectionsState led to a subset
of test testing invalid invariants, for example when we have at the same
time an available connecion and a waiter, which is an invalid state of
the system.

Modifications:

* test of ConnectionsState
* ConnectionsState
* ConnectionPool

are modified so the state under tests is achieved only by a sequence of
modifications invoked by state API.

During modification some tests are eliminated as they were testing
artificial state, which can not be achieved by exposed APIs.

ConnectionsState is pruned from "replace" as in no valid state we can
have a situation when we can "replace" a connection.

Invalid invariants and tests are removed.

Result:

We do not have a way to modify state of the ConnectionsState by direct
interaction with private state of the object. Getter on the state is
considered harmless and used for tests only.
diff --git a/Sources/AsyncHTTPClient/ConnectionPool.swift b/Sources/AsyncHTTPClient/ConnectionPool.swift
@@ -235,7 +235,7 @@ class HTTP1ConnectionProvider {
                     logger.trace("opening fresh connection (found matching but inactive connection)",
                                  metadata: ["ahc-dead-connection": "\(connection)"])
                     self.makeChannel(preference: waiter.preference).whenComplete { result in
-                        self.connect(result, waiter: waiter, replacing: connection, logger: logger)
+                        self.connect(result, waiter: waiter, logger: logger)
                     }
                 }
             }
@@ -252,7 +252,7 @@ class HTTP1ConnectionProvider {
                              metadata: ["ahc-old-connection": "\(connection)",
                                         "ahc-waiter": "\(waiter)"])
                 self.makeChannel(preference: waiter.preference).whenComplete { result in
-                    self.connect(result, waiter: waiter, replacing: connection, logger: logger)
+                    self.connect(result, waiter: waiter, logger: logger)
                 }
             }
         case .park(let connection):
@@ -308,21 +308,15 @@ class HTTP1ConnectionProvider {
         return waiter.promise.futureResult
     }
 
-    func connect(_ result: Result<Channel, Error>,
-                 waiter: Waiter<Connection>,
-                 replacing closedConnection: Connection? = nil,
-                 logger: Logger) {
+    func connect(_ result: Result<Channel, Error>, waiter: Waiter<Connection>, logger: Logger) {
         let action: Action<Connection>
         switch result {
         case .success(let channel):
             logger.trace("successfully created connection",
                          metadata: ["ahc-connection": "\(channel)"])
             let connection = Connection(channel: channel, provider: self)
             action = self.lock.withLock {
-                if let closedConnection = closedConnection {
-                    self.state.drop(connection: closedConnection)
-                }
-                return self.state.offer(connection: connection)
+                self.state.offer(connection: connection)
             }
 
             switch action {
@@ -367,7 +361,7 @@ class HTTP1ConnectionProvider {
             // This is needed to start a new stack, otherwise, since this is called on a previous
             // future completion handler chain, it will be growing indefinitely until the connection is closed.
             // We might revisit this when https://github.com/apple/swift-nio/issues/970 is resolved.
-            connection.channel.eventLoop.execute {
+            connection.eventLoop.execute {
                 self.execute(action, logger: logger)
             }
         }
@@ -418,59 +412,7 @@ class HTTP1ConnectionProvider {
     }
 
     private func makeChannel(preference: HTTPClient.EventLoopPreference) -> EventLoopFuture<Channel> {
-        let channelEventLoop = preference.bestEventLoop ?? self.eventLoop
-        let requiresTLS = self.key.scheme.requiresTLS
-        let bootstrap: NIOClientTCPBootstrap
-        do {
-            bootstrap = try NIOClientTCPBootstrap.makeHTTPClientBootstrapBase(on: channelEventLoop, host: self.key.host, port: self.key.port, requiresTLS: requiresTLS, configuration: self.configuration)
-        } catch {
-            return channelEventLoop.makeFailedFuture(error)
-        }
-
-        let channel: EventLoopFuture<Channel>
-        switch self.key.scheme {
-        case .http, .https:
-            let address = HTTPClient.resolveAddress(host: self.key.host, port: self.key.port, proxy: self.configuration.proxy)
-            channel = bootstrap.connect(host: address.host, port: address.port)
-        case .unix, .http_unix, .https_unix:
-            channel = bootstrap.connect(unixDomainSocketPath: self.key.unixPath)
-        }
-
-        return channel.flatMap { channel in
-            let requiresSSLHandler = self.configuration.proxy != nil && self.key.scheme.requiresTLS
-            let handshakePromise = channel.eventLoop.makePromise(of: Void.self)
-
-            channel.pipeline.addSSLHandlerIfNeeded(for: self.key, tlsConfiguration: self.configuration.tlsConfiguration, addSSLClient: requiresSSLHandler, handshakePromise: handshakePromise)
-
-            return handshakePromise.futureResult.flatMap {
-                channel.pipeline.addHTTPClientHandlers(leftOverBytesStrategy: .forwardBytes)
-            }.flatMap {
-                #if canImport(Network)
-                    if #available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *), bootstrap.underlyingBootstrap is NIOTSConnectionBootstrap {
-                        return channel.pipeline.addHandler(HTTPClient.NWErrorHandler(), position: .first)
-                    }
-                #endif
-                return channel.eventLoop.makeSucceededFuture(())
-            }.flatMap {
-                switch self.configuration.decompression {
-                case .disabled:
-                    return channel.eventLoop.makeSucceededFuture(())
-                case .enabled(let limit):
-                    let decompressHandler = NIOHTTPResponseDecompressor(limit: limit)
-                    return channel.pipeline.addHandler(decompressHandler)
-                }
-            }.map {
-                channel
-            }
-        }.flatMapError { error in
-            #if canImport(Network)
-                var error = error
-                if #available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *), bootstrap.underlyingBootstrap is NIOTSConnectionBootstrap {
-                    error = HTTPClient.NWErrorHandler.translateError(error)
-                }
-            #endif
-            return channelEventLoop.makeFailedFuture(error)
-        }
+        return NIOClientTCPBootstrap.makeHTTP1Channel(destination: self.key, eventLoop: self.eventLoop, configuration: self.configuration, preference: preference)
     }
 
     /// A `Waiter` represents a request that waits for a connection when none is
diff --git a/Sources/AsyncHTTPClient/ConnectionsState.swift b/Sources/AsyncHTTPClient/ConnectionsState.swift
@@ -74,15 +74,6 @@ extension HTTP1ConnectionProvider {
             return Snapshot(state: self.state, availableConnections: self.availableConnections, leasedConnections: self.leasedConnections, waiters: self.waiters, openedConnectionsCount: self.openedConnectionsCount, pending: self.pending)
         }
 
-        mutating func testsOnly_setInternalState(_ snapshot: Snapshot<ConnectionType>) {
-            self.state = snapshot.state
-            self.availableConnections = snapshot.availableConnections
-            self.leasedConnections = snapshot.leasedConnections
-            self.waiters = snapshot.waiters
-            self.openedConnectionsCount = snapshot.openedConnectionsCount
-            self.pending = snapshot.pending
-        }
-
         func assertInvariants() {
             assert(self.waiters.isEmpty)
             assert(self.availableConnections.isEmpty)
@@ -158,30 +149,22 @@ extension HTTP1ConnectionProvider {
 
                 if connection.isActiveEstimation, !closing { // If connection is alive, we can offer it to a next waiter
                     if let waiter = self.waiters.popFirst() {
+                        // There should be no case where we have both capacity and a waiter here.
+                        // Waiter can only exists if there was no capacity at aquire. If some connection
+                        // is released when we have waiter it can only indicate that we should lease (if EL are the same),
+                        // or replace (if they are different). But we cannot increase connection count here.
+                        assert(!self.hasCapacity)
+
                         let (eventLoop, required) = self.resolvePreference(waiter.preference)
 
                         // If returned connection is on same EL or we do not require special EL - lease it
                         if connection.eventLoop === eventLoop || !required {
                             return .lease(connection, waiter)
                         }
 
-                        // If there is an opened connection on the same loop, lease it and park returned
-                        if let found = self.availableConnections.firstIndex(where: { $0.eventLoop === eventLoop }) {
-                            self.leasedConnections.remove(ConnectionKey(connection))
-                            let replacement = self.availableConnections.swap(at: found, with: connection)
-                            self.leasedConnections.insert(ConnectionKey(replacement))
-                            return .parkAnd(connection, .lease(replacement, waiter))
-                        }
-
-                        // If we can create new connection - do it
-                        if self.hasCapacity {
-                            self.leasedConnections.remove(ConnectionKey(connection))
-                            self.availableConnections.append(connection)
-                            self.openedConnectionsCount += 1
-                            return .parkAnd(connection, .create(waiter))
-                        }
-
                         // If we cannot create new connections, we will have to replace returned connection with a new one on the required loop
+                        // We will keep the `openedConnectionCount`, since .replace === .create, so we decrease and increase the `openedConnectionCount`
+                        self.leasedConnections.remove(ConnectionKey(connection))
                         return .replace(connection, waiter)
                     } else { // or park, if there are no waiters
                         self.leasedConnections.remove(ConnectionKey(connection))
@@ -214,15 +197,6 @@ extension HTTP1ConnectionProvider {
             }
         }
 
-        mutating func drop(connection: ConnectionType) {
-            switch self.state {
-            case .active:
-                self.leasedConnections.remove(ConnectionKey(connection))
-            case .closed:
-                assertionFailure("should not happen")
-            }
-        }
-
         mutating func connectFailed() -> Action<ConnectionType> {
             switch self.state {
             case .active:
@@ -287,20 +261,25 @@ extension HTTP1ConnectionProvider {
 
         mutating func processNextWaiter() -> Action<ConnectionType> {
             if let waiter = self.waiters.popFirst() {
-                let (eventLoop, required) = self.resolvePreference(waiter.preference)
-
-                // If specific EL is required, we have only two options - find open one or create a new one
-                if required, let found = self.availableConnections.firstIndex(where: { $0.eventLoop === eventLoop }) {
-                    let connection = self.availableConnections.remove(at: found)
-                    self.leasedConnections.insert(ConnectionKey(connection))
-                    return .lease(connection, waiter)
-                } else if !required, let connection = self.availableConnections.popFirst() {
-                    self.leasedConnections.insert(ConnectionKey(connection))
-                    return .lease(connection, waiter)
-                } else {
-                    self.openedConnectionsCount += 1
-                    return .create(waiter)
-                }
+                // There should be no case where we have waiters and available connections at the same time.
+                //
+                // This method is called in following cases:
+                //
+                // 1. from `release` when connection is inactive and cannot be re-used
+                // 2. from `connectFailed` when we failed to establish a new connection
+                // 3. from `remoteClose` when connection was closed by the remote side and cannot be re-used
+                // 4. from `timeout` when connection was closed due to idle timeout and cannot be re-used.
+                //
+                // In all cases connection, which triggered the transition, will not be in `available` state.
+                //
+                // Given that the waiter can only be present in the pool if there were no available connections
+                // (otherwise it had been leased a connection immediately on getting the connection), we do not
+                // see a situation when we can lease another available connection, therefore the only course
+                // of action is to create a new connection for the waiter.
+                assert(self.availableConnections.isEmpty)
+
+                self.openedConnectionsCount += 1
+                return .create(waiter)
             }
 
             // if capacity is at max and the are no waiters and no in-flight requests for connection, we are closing this provider
diff --git a/Sources/AsyncHTTPClient/Utils.swift b/Sources/AsyncHTTPClient/Utils.swift
@@ -138,6 +138,65 @@ extension NIOClientTCPBootstrap {
                 return channelAddedFuture
             }
     }
+
+    static func makeHTTP1Channel(destination: ConnectionPool.Key, eventLoop: EventLoop, configuration: HTTPClient.Configuration, preference: HTTPClient.EventLoopPreference) -> EventLoopFuture<Channel> {
+        let channelEventLoop = preference.bestEventLoop ?? eventLoop
+
+        let key = destination
+
+        let requiresTLS = key.scheme.requiresTLS
+        let bootstrap: NIOClientTCPBootstrap
+        do {
+            bootstrap = try NIOClientTCPBootstrap.makeHTTPClientBootstrapBase(on: channelEventLoop, host: key.host, port: key.port, requiresTLS: requiresTLS, configuration: configuration)
+        } catch {
+            return channelEventLoop.makeFailedFuture(error)
+        }
+
+        let channel: EventLoopFuture<Channel>
+        switch key.scheme {
+        case .http, .https:
+            let address = HTTPClient.resolveAddress(host: key.host, port: key.port, proxy: configuration.proxy)
+            channel = bootstrap.connect(host: address.host, port: address.port)
+        case .unix, .http_unix, .https_unix:
+            channel = bootstrap.connect(unixDomainSocketPath: key.unixPath)
+        }
+
+        return channel.flatMap { channel in
+            let requiresSSLHandler = configuration.proxy != nil && key.scheme.requiresTLS
+            let handshakePromise = channel.eventLoop.makePromise(of: Void.self)
+
+            channel.pipeline.addSSLHandlerIfNeeded(for: key, tlsConfiguration: configuration.tlsConfiguration, addSSLClient: requiresSSLHandler, handshakePromise: handshakePromise)
+
+            return handshakePromise.futureResult.flatMap {
+                channel.pipeline.addHTTPClientHandlers(leftOverBytesStrategy: .forwardBytes)
+            }.flatMap {
+                #if canImport(Network)
+                    if #available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *), bootstrap.underlyingBootstrap is NIOTSConnectionBootstrap {
+                        return channel.pipeline.addHandler(HTTPClient.NWErrorHandler(), position: .first)
+                    }
+                #endif
+                return channel.eventLoop.makeSucceededFuture(())
+            }.flatMap {
+                switch configuration.decompression {
+                case .disabled:
+                    return channel.eventLoop.makeSucceededFuture(())
+                case .enabled(let limit):
+                    let decompressHandler = NIOHTTPResponseDecompressor(limit: limit)
+                    return channel.pipeline.addHandler(decompressHandler)
+                }
+            }.map {
+                channel
+            }
+        }.flatMapError { error in
+            #if canImport(Network)
+                var error = error
+                if #available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *), bootstrap.underlyingBootstrap is NIOTSConnectionBootstrap {
+                    error = HTTPClient.NWErrorHandler.translateError(error)
+                }
+            #endif
+            return channelEventLoop.makeFailedFuture(error)
+        }
+    }
 }
 
 extension Connection {
diff --git a/Tests/AsyncHTTPClientTests/ConnectionPoolTests+XCTest.swift b/Tests/AsyncHTTPClientTests/ConnectionPoolTests+XCTest.swift
@@ -41,18 +41,10 @@ extension ConnectionPoolTests {
             ("testReleaseInactiveConnectionEmptyQueueHasConnections", testReleaseInactiveConnectionEmptyQueueHasConnections),
             ("testReleaseAliveConnectionHasWaiter", testReleaseAliveConnectionHasWaiter),
             ("testReleaseInactiveConnectionHasWaitersNoConnections", testReleaseInactiveConnectionHasWaitersNoConnections),
-            ("testReleaseInactiveConnectionHasWaitersHasConnections", testReleaseInactiveConnectionHasWaitersHasConnections),
             ("testReleaseAliveConnectionSameELHasWaiterSpecificEL", testReleaseAliveConnectionSameELHasWaiterSpecificEL),
-            ("testReleaseAliveConnectionDifferentELNoSameELConnectionsHasWaiterSpecificEL", testReleaseAliveConnectionDifferentELNoSameELConnectionsHasWaiterSpecificEL),
-            ("testReleaseAliveConnectionDifferentELHasSameELConnectionsHasWaiterSpecificEL", testReleaseAliveConnectionDifferentELHasSameELConnectionsHasWaiterSpecificEL),
             ("testReleaseAliveConnectionDifferentELNoSameELConnectionsOnLimitHasWaiterSpecificEL", testReleaseAliveConnectionDifferentELNoSameELConnectionsOnLimitHasWaiterSpecificEL),
-            ("testReleaseInactiveConnectionHasWaitersHasSameELConnectionsSpecificEL", testReleaseInactiveConnectionHasWaitersHasSameELConnectionsSpecificEL),
-            ("testReleaseInactiveConnectionHasWaitersNoSameELConnectionsSpecificEL", testReleaseInactiveConnectionHasWaitersNoSameELConnectionsSpecificEL),
             ("testNextWaiterEmptyQueue", testNextWaiterEmptyQueue),
             ("testNextWaiterEmptyQueueHasConnections", testNextWaiterEmptyQueueHasConnections),
-            ("testNextWaiterHasWaitersHasConnections", testNextWaiterHasWaitersHasConnections),
-            ("testNextWaiterHasWaitersHasSameELConnectionsSpecificEL", testNextWaiterHasWaitersHasSameELConnectionsSpecificEL),
-            ("testNextWaiterHasWaitersHasDifferentELConnectionsSpecificEL", testNextWaiterHasWaitersHasDifferentELConnectionsSpecificEL),
             ("testTimeoutLeasedConnection", testTimeoutLeasedConnection),
             ("testTimeoutAvailableConnection", testTimeoutAvailableConnection),
             ("testRemoteClosedLeasedConnection", testRemoteClosedLeasedConnection),
diff --git a/Tests/AsyncHTTPClientTests/ConnectionPoolTests.swift b/Tests/AsyncHTTPClientTests/ConnectionPoolTests.swift
diff --git a/Tests/AsyncHTTPClientTests/ConnectionPoolTestsSupport.swift b/Tests/AsyncHTTPClientTests/ConnectionPoolTestsSupport.swift
