diff --git a/Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift b/Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift index 676df915a..acb824277 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift +++ b/Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift @@ -77,6 +77,8 @@ final class HTTPConnectionPool: idGenerator: idGenerator, maximumConcurrentHTTP1Connections: clientConfiguration.connectionPool .concurrentHTTP1ConnectionsPerHostSoftLimit, + maximumConcurrentHTTP2Connections: clientConfiguration.connectionPool + .concurrentHTTP2ConnectionsPerHostSoftLimit, retryConnectionEstablishment: clientConfiguration.connectionPool.retryConnectionEstablishment, preferHTTP1: clientConfiguration.httpVersion == .http1Only, maximumConnectionUses: clientConfiguration.maximumUsesPerConnection, diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift index 2a0e0cc80..955306b8c 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift +++ b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift @@ -522,6 +522,28 @@ extension HTTPConnectionPool { } } + /// Prefer existing stream capacity and allow only one pending connection attempt for the + /// requests being considered. A missing required event loop may exceed the pool's soft limit. + func canCreateConnection(onRequired eventLoop: EventLoop?, maximumConcurrentConnections: Int) -> Bool { + var connectionCount = 0 + var hasEligibleConnection = false + var hasAvailableOrPendingConnection = false + for connection in self.connections where connection.canOrWillBeAbleToExecuteRequests { + connectionCount += 1 + guard eventLoop == nil || connection.eventLoop === eventLoop else { continue } + hasEligibleConnection = true + if connection.isAvailable || connection.isStartingOrBackingOff { + hasAvailableOrPendingConnection = true + } + } + return !hasEligibleConnection + || (connectionCount < maximumConcurrentConnections && !hasAvailableOrPendingConnection) + } + + func hasReachedActiveConnectionLimit(_ limit: Int) -> Bool { + self.connections.lazy.filter { $0.isActive }.count >= limit + } + /// used after backoff is done to determine if we need to create a new connection /// - Parameters: /// - eventLoop: connection `EventLoop` to search for @@ -545,11 +567,6 @@ extension HTTPConnectionPool { } mutating func createNewConnection(on eventLoop: EventLoop) -> Connection.ID { - assert( - !self.hasConnectionThatCanOrWillBeAbleToExecuteRequests(for: eventLoop), - "we should not create more than one connection per event loop" - ) - let connection = HTTP2ConnectionState( connectionID: self.generator.next(), eventLoop: eventLoop, diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift index 67a07e6dd..540b6168d 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift +++ b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift @@ -37,13 +37,17 @@ extension HTTPConnectionPool { /// The property was introduced to fail fast during testing. /// Otherwise this should always be true and not turned off. private let retryConnectionEstablishment: Bool + private let maximumConcurrentConnections: Int init( idGenerator: Connection.ID.Generator, retryConnectionEstablishment: Bool, lifecycleState: StateMachine.LifecycleState, - maximumConnectionUses: Int? + maximumConnectionUses: Int?, + maximumConcurrentConnections: Int = 1 ) { + precondition(maximumConcurrentConnections > 0) + self.maximumConcurrentConnections = maximumConcurrentConnections self.idGenerator = idGenerator self.requests = RequestQueue() @@ -62,7 +66,7 @@ extension HTTPConnectionPool { newHTTP2Connection: Connection, maxConcurrentStreams: Int ) -> Action { - let migrationAction = self.migrateConnectionsAndRequestsFromHTTP1( + var migrationAction = self.migrateConnectionsAndRequestsFromHTTP1( http1Connections: http1Connections, http2Connections: http2Connections, requests: requests @@ -73,6 +77,16 @@ extension HTTPConnectionPool { maxConcurrentStreams: maxConcurrentStreams ) + // The shared migration combiner discards HTTP/1 prewarming actions. HTTP/2 growth, + // however, represents queued demand and must survive the protocol migration. + switch newConnectionAction.connection { + case .createConnection(let connectionID, let eventLoop), + .scheduleTimeoutTimerAndCreateConnection(_, let connectionID, let eventLoop): + migrationAction.createConnections.append((connectionID, eventLoop)) + default: + break + } + return .init( request: newConnectionAction.request, connection: .combined(migrationAction, newConnectionAction.connection) @@ -163,14 +177,17 @@ extension HTTPConnectionPool { /// 2. No available stream so we definitely need to wait until we have one self.requests.push(request) - if self.connections.hasConnectionThatCanOrWillBeAbleToExecuteRequests(for: eventLoop) { - /// 3. we already have a connection, we just need to wait until until it becomes available + if !self.connections.canCreateConnection( + onRequired: eventLoop, + maximumConcurrentConnections: self.maximumConcurrentConnections + ) { + // Wait for an existing connection or for capacity within the soft limit. return .init( request: .scheduleRequestTimeout(for: request, on: eventLoop), connection: .none ) } else { - /// 4. we do *not* have a connection, need to create a new one and wait until it is connected. + // Establish missing event loop coverage or expand a saturated pool. let connectionId = self.connections.createNewConnection(on: eventLoop) return .init( request: .scheduleRequestTimeout(for: request, on: eventLoop), @@ -200,14 +217,17 @@ extension HTTPConnectionPool { /// 2. No available stream so we definitely need to wait until we have one self.requests.push(request) - if self.connections.hasConnectionThatCanOrWillBeAbleToExecuteRequests { - /// 3. we already have a connection, we just need to wait until until it becomes available + if !self.connections.canCreateConnection( + onRequired: nil, + maximumConcurrentConnections: self.maximumConcurrentConnections + ) { + // Wait for an existing connection or for capacity within the soft limit. return .init( request: .scheduleRequestTimeout(for: request, on: eventLoop), connection: .none ) } else { - /// 4. we do *not* have a connection, need to create a new one and wait until it is connected. + // Establish the first connection or expand a saturated pool. let connectionId = self.connections.createNewConnection(on: eventLoop) return .init( request: .scheduleRequestTimeout(for: request, on: eventLoop), @@ -227,11 +247,17 @@ extension HTTPConnectionPool { self.failedConsecutiveConnectionAttempts = 0 self.lastConnectFailure = nil let doesConnectionExistsForEL = self.connections.hasActiveConnection(for: connection.eventLoop) + let reachedLimit = self.connections.hasReachedActiveConnectionLimit(self.maximumConcurrentConnections) + // Preserve the default's historical migration behavior. With a larger limit, keep + // connections beyond the limit only when required event loop requests need them. + let needsEventLoopCoverage = + !doesConnectionExistsForEL + && (self.maximumConcurrentConnections == 1 || !self.requests.isEmpty(for: connection.eventLoop)) let (index, context) = self.connections.newHTTP2ConnectionEstablished( connection, maxConcurrentStreams: maxConcurrentStreams ) - if doesConnectionExistsForEL { + if reachedLimit && !needsEventLoopCoverage { let connection = self.connections.closeConnection(at: index) return .init( request: .none, @@ -270,8 +296,18 @@ extension HTTPConnectionPool { }() let connectionAction = { () -> EstablishedConnectionAction in + let newConnectionID = self.createConnectionForPendingRequests(on: context.eventLoop) if context.isIdle, requestsToExecute.isEmpty { + if let newConnectionID = newConnectionID { + return .scheduleTimeoutTimerAndCreateConnection( + timeoutID: context.connectionID, + newConnectionID: newConnectionID, + on: context.eventLoop + ) + } return .scheduleTimeoutTimer(context.connectionID, on: context.eventLoop) + } else if let newConnectionID = newConnectionID { + return .createConnection(connectionID: newConnectionID, on: context.eventLoop) } else { return .none } @@ -341,41 +377,30 @@ extension HTTPConnectionPool { private mutating func nextActionForFailedConnection(at index: Int, on eventLoop: EventLoop) -> Action { switch self.lifecycleState { case .running: - // 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. - - // precompute if we have starting or active connections to only iterate once over `self.connections` + // Preserve retry ordering when no starting or active connection can serve the + // requests. Other backoff timers must not delay replacing this connection. let context = self.connections.backingOffTimerDone(for: eventLoop) - - // we need to start a new on connection in two cases: - let needGeneralPurposeConnection = - // 1. if we have general purpose requests + let needsGeneralPurposeConnection = !self.requests.isEmpty(for: nil) - // and no connection starting or active && !context.hasGeneralPurposeConnection - - let needRequiredEventLoopConnection = - // 2. or if we have requests for a required event loop + let needsRequiredConnection = !self.requests.isEmpty(for: eventLoop) - // and no connection starting or active for the given event loop && !context.hasConnectionOnSpecifiedEventLoop + if needsGeneralPurposeConnection || needsRequiredConnection { + let (connectionID, eventLoop) = self.connections + .createNewConnectionByReplacingClosedConnection(at: index) + return .init(request: .none, connection: .createConnection(connectionID, on: eventLoop)) + } - guard needGeneralPurposeConnection || needRequiredEventLoopConnection else { - // otherwise we can remove the connection - self.connections.removeConnection(at: index) + self.connections.removeConnection(at: index) + guard let (newConnectionID, newEventLoop) = self.createConnectionAfterCapacityIsReleased(on: eventLoop) + else { return .none } - let (newConnectionID, previousEventLoop) = self.connections - .createNewConnectionByReplacingClosedConnection(at: index) - precondition(previousEventLoop === eventLoop) - return .init( request: .none, - connection: .createConnection(newConnectionID, on: eventLoop) + connection: .createConnection(newConnectionID, on: newEventLoop) ) case .shuttingDown(let unclean): @@ -397,22 +422,57 @@ extension HTTPConnectionPool { private mutating func nextActionForClosingConnection(on eventLoop: EventLoop) -> Action { switch self.lifecycleState { case .running: - let hasPendingRequest = !self.requests.isEmpty(for: eventLoop) || !self.requests.isEmpty(for: nil) - guard hasPendingRequest else { + guard let (newConnectionID, newEventLoop) = self.createConnectionAfterCapacityIsReleased(on: eventLoop) + else { return .none } - let newConnectionID = self.connections.createNewConnection(on: eventLoop) - return .init( request: .none, - connection: .createConnection(newConnectionID, on: eventLoop) + connection: .createConnection(newConnectionID, on: newEventLoop) ) case .shutDown, .shuttingDown: return .none } } + private mutating func createConnectionForPendingRequests(on eventLoop: EventLoop) -> Connection.ID? { + let needsRequiredConnection = + !self.requests.isEmpty(for: eventLoop) + && self.connections.canCreateConnection( + onRequired: eventLoop, + maximumConcurrentConnections: self.maximumConcurrentConnections + ) + let needsGeneralPurposeConnection = + !self.requests.isEmpty(for: nil) + && self.connections.canCreateConnection( + onRequired: nil, + maximumConcurrentConnections: self.maximumConcurrentConnections + ) + guard needsRequiredConnection || needsGeneralPurposeConnection else { return nil } + return self.connections.createNewConnection(on: eventLoop) + } + + private mutating func createConnectionAfterCapacityIsReleased( + on eventLoop: EventLoop + ) -> (Connection.ID, EventLoop)? { + if let connectionID = self.createConnectionForPendingRequests(on: eventLoop) { + return (connectionID, eventLoop) + } + // The released slot belongs to the whole pool. Requests on another required event + // loop must not wait for a stream to close there before they can use this capacity. + for requiredEventLoop in self.requests.eventLoopsWithPendingRequests() where requiredEventLoop !== eventLoop + { + if self.connections.canCreateConnection( + onRequired: requiredEventLoop, + maximumConcurrentConnections: self.maximumConcurrentConnections + ) { + return (self.connections.createNewConnection(on: requiredEventLoop), requiredEventLoop) + } + } + return nil + } + mutating func http2ConnectionStreamClosed(_ connectionID: Connection.ID) -> Action { let (index, context) = self.connections.releaseStream(connectionID) return .init(self.nextActionForAvailableConnection(at: index, context: context)) @@ -531,6 +591,18 @@ extension HTTPConnectionPool { "If we are shutting down, we must not have any idle connections" ) + if let (newConnectionID, eventLoop) = self.createConnectionAfterCapacityIsReleased(on: connection.eventLoop) + { + return .init( + request: .none, + connection: .closeConnectionAndCreateConnection( + closeConnection: connection, + newConnectionID: newConnectionID, + on: eventLoop + ) + ) + } + return .init( request: .none, connection: .closeConnection(connection, isShutdown: .no) diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+StateMachine.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+StateMachine.swift index b905e26bd..95e001e55 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+StateMachine.swift +++ b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+StateMachine.swift @@ -113,6 +113,7 @@ extension HTTPConnectionPool { let idGenerator: Connection.ID.Generator let maximumConcurrentHTTP1Connections: Int + let maximumConcurrentHTTP2Connections: Int /// The property was introduced to fail fast during testing. /// Otherwise this should always be true and not turned off. private let retryConnectionEstablishment: Bool @@ -122,12 +123,14 @@ extension HTTPConnectionPool { init( idGenerator: Connection.ID.Generator, maximumConcurrentHTTP1Connections: Int, + maximumConcurrentHTTP2Connections: Int = 1, retryConnectionEstablishment: Bool, preferHTTP1: Bool, maximumConnectionUses: Int?, preWarmedHTTP1ConnectionCount: Int ) { self.maximumConcurrentHTTP1Connections = maximumConcurrentHTTP1Connections + self.maximumConcurrentHTTP2Connections = maximumConcurrentHTTP2Connections self.retryConnectionEstablishment = retryConnectionEstablishment self.idGenerator = idGenerator self.maximumConnectionUses = maximumConnectionUses @@ -148,7 +151,8 @@ extension HTTPConnectionPool { idGenerator: idGenerator, retryConnectionEstablishment: retryConnectionEstablishment, lifecycleState: .running, - maximumConnectionUses: maximumConnectionUses + maximumConnectionUses: maximumConnectionUses, + maximumConcurrentConnections: maximumConcurrentHTTP2Connections ) self.state = .http2(http2State) } @@ -201,7 +205,8 @@ extension HTTPConnectionPool { idGenerator: self.idGenerator, retryConnectionEstablishment: self.retryConnectionEstablishment, lifecycleState: http1StateMachine.lifecycleState, - maximumConnectionUses: self.maximumConnectionUses + maximumConnectionUses: self.maximumConnectionUses, + maximumConcurrentConnections: self.maximumConcurrentHTTP2Connections ) let migrationAction = http2StateMachine.migrateFromHTTP1( http1Connections: http1StateMachine.connections, diff --git a/Sources/AsyncHTTPClient/HTTPClient.swift b/Sources/AsyncHTTPClient/HTTPClient.swift index cc8792497..fcfdbe534 100644 --- a/Sources/AsyncHTTPClient/HTTPClient.swift +++ b/Sources/AsyncHTTPClient/HTTPClient.swift @@ -1409,6 +1409,20 @@ extension HTTPClient.Configuration { /// an explicit eventLoopRequirement are sent, this number might be exceeded due to overflow connections. public var concurrentHTTP1ConnectionsPerHostSoftLimit: Int = 8 + /// The maximum number of HTTP/2 connections per host that can accept new requests. + /// + /// Connections are created on demand when existing connections have no available streams, + /// up to this limit, waiting for a pending attempt before expanding further. Starting and + /// backing-off connections count towards the limit; draining connections do not. + /// Requests with an explicit event loop requirement can exceed the limit + /// to establish a connection on their required event loop. With the default limit, protocol + /// migration can also retain one connection per event loop. Defaults to `1` and must be positive. + public var concurrentHTTP2ConnectionsPerHostSoftLimit: Int = 1 { + didSet { + precondition(self.concurrentHTTP2ConnectionsPerHostSoftLimit > 0) + } + } + /// If true, ``HTTPClient`` will try to create new connections on connection failure with an exponential backoff. /// Requests will only fail after the ``HTTPClient/Configuration/Timeout-swift.struct/connect`` timeout exceeded. /// If false, all requests that have no assigned connection will fail immediately after a connection could not be established. diff --git a/Sources/AsyncHTTPClient/HTTPClientConfiguration+SwiftConfiguration.swift b/Sources/AsyncHTTPClient/HTTPClientConfiguration+SwiftConfiguration.swift index 9430689e9..a014925d9 100644 --- a/Sources/AsyncHTTPClient/HTTPClientConfiguration+SwiftConfiguration.swift +++ b/Sources/AsyncHTTPClient/HTTPClientConfiguration+SwiftConfiguration.swift @@ -124,6 +124,7 @@ extension HTTPClient.Configuration.ConnectionPool { /// ## Configuration keys: /// - `idleTimeoutMs` (int, optional, default: 60,000): Connection idle timeout in milliseconds. /// - `concurrentHTTP1ConnectionsPerHostSoftLimit` (int, optional, default: 8): Soft limit for concurrent HTTP/1.1 connections per host. + /// - `concurrentHTTP2ConnectionsPerHostSoftLimit` (positive int, optional, default: 1): Soft limit for concurrent HTTP/2 connections per host. /// - `retryConnectionEstablishment` (bool, optional, default: true): Retry failed connection establishment. /// - `preWarmedHTTP1ConnectionCount` (int, optional, default: 0): Number of pre-warmed HTTP/1.1 connections per host. public init(configReader: ConfigReader) { @@ -133,6 +134,10 @@ extension HTTPClient.Configuration.ConnectionPool { forKey: "concurrentHTTP1ConnectionsPerHostSoftLimit", default: 8 ) + self.concurrentHTTP2ConnectionsPerHostSoftLimit = configReader.int( + forKey: "concurrentHTTP2ConnectionsPerHostSoftLimit", + default: 1 + ) self.retryConnectionEstablishment = configReader.bool(forKey: "retryConnectionEstablishment", default: true) self.preWarmedHTTP1ConnectionCount = configReader.int(forKey: "preWarmedHTTP1ConnectionCount", default: 0) } diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index 183a227bd..db90ea3b6 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -113,6 +113,40 @@ class HTTP2ClientTests: XCTestCase { XCTAssertNoThrow(try EventLoopFuture.whenAllComplete(requestPromises, on: el).wait()) } + func testHTTP2ConnectionSoftLimitAllowsConcurrentRequestsBeyondServerStreamLimit() throws { + let bin = HTTPBin(.http2(settings: [.init(parameter: .maxConcurrentStreams, value: 1)])) { _ in + SendHeaderAndWaitChannelHandler() + } + defer { XCTAssertNoThrow(try bin.shutdown()) } + let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) + defer { XCTAssertNoThrow(try group.syncShutdownGracefully()) } + var configuration = HTTPClient.Configuration() + configuration.tlsConfiguration = .clientDefault + configuration.tlsConfiguration?.certificateVerification = .none + configuration.connectionPool.concurrentHTTP2ConnectionsPerHostSoftLimit = 2 + let client = HTTPClient(eventLoopGroupProvider: .shared(group), configuration: configuration) + defer { XCTAssertNoThrow(try client.syncShutdown()) } + + // The server leaves responses open, so each received head proves one concurrently + // executing request. Two heads therefore require two connections with this SETTINGS limit. + let receivedHeads = self.expectation(description: "Two requests execute concurrently") + receivedHeads.expectedFulfillmentCount = 2 + let request = try HTTPClient.Request(url: "https://localhost:\(bin.port)/wait") + let tasks = (0..<3).map { _ in + client.execute( + request: request, + delegate: HeadReceivedCallback { head in + XCTAssertEqual(head.version, .http2) + receivedHeads.fulfill() + } + ) + } + self.wait(for: [receivedHeads], timeout: 5) + XCTAssertEqual(bin.createdConnections, 2) + // Keep all tasks alive until shutdown; cancelling one early would free a stream for the third. + withExtendedLifetime(tasks) {} + } + func testConcurrentRequestsFromDifferentThreads() { let bin = HTTPBin(.http2(compress: false)) defer { XCTAssertNoThrow(try bin.shutdown()) } diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2SoftLimitTests.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2SoftLimitTests.swift new file mode 100644 index 000000000..aa3dda44f --- /dev/null +++ b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2SoftLimitTests.swift @@ -0,0 +1,357 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the AsyncHTTPClient open source project +// +// Copyright (c) 2026 Apple Inc. and the AsyncHTTPClient project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of AsyncHTTPClient project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +import NIOCore +import NIOEmbedded +import XCTest + +@testable import AsyncHTTPClient + +final class HTTPConnectionPool_HTTP2SoftLimitTests: XCTestCase { + private typealias State = HTTPConnectionPool.StateMachine + private typealias Connection = HTTPConnectionPool.Connection + + private func makeState(limit: Int = 2, preferHTTP1: Bool = false, http1Limit: Int = 8) -> State { + State( + idGenerator: .init(), + maximumConcurrentHTTP1Connections: http1Limit, + maximumConcurrentHTTP2Connections: limit, + retryConnectionEstablishment: true, + preferHTTP1: preferHTTP1, + maximumConnectionUses: nil, + preWarmedHTTP1ConnectionCount: 0 + ) + } + + private func request(on eventLoop: EventLoop, required: Bool = false) -> HTTPConnectionPool.Request { + HTTPConnectionPool.Request( + MockHTTPScheduableRequest(eventLoop: eventLoop, requiresEventLoopForChannel: required) + ) + } + + private func createdConnection( + _ action: State.Action, + file: StaticString = #filePath, + line: UInt = #line + ) throws -> Connection { + guard case .createConnection(let id, let eventLoop) = action.connection else { + XCTFail("Expected connection creation, got \(action.connection)", file: file, line: line) + throw HTTPClientError.connectTimeout + } + return .__testOnly_connection(id: id, eventLoop: eventLoop) + } + + func testSaturatedConnectionExpandsButRespectsSoftLimit() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState() + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + + let secondRequest = self.request(on: eventLoop) + let second = try self.createdConnection(state.executeRequest(secondRequest)) + XCTAssertEqual(state.executeRequest(self.request(on: eventLoop)).connection, .none) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1).request, + .executeRequestsAndCancelTimeouts([secondRequest], second) + ) + XCTAssertEqual(state.executeRequest(self.request(on: eventLoop)).connection, .none) + } + + func testInitialQueuedBurstExpandsAfterSettingsAndReusesAvailableStreams() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState() + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + let queued = (0..<4).map { _ in self.request(on: eventLoop) } + for request in queued { + XCTAssertEqual(state.executeRequest(request).connection, .none) + } + + let established = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 2) + let second = try self.createdConnection(established) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 4).request, + .executeRequestsAndCancelTimeouts(Array(queued.suffix(3)), second) + ) + let nextRequest = self.request(on: eventLoop) + XCTAssertEqual( + state.executeRequest(nextRequest).request, + .executeRequest(nextRequest, second, cancelTimeout: false) + ) + XCTAssertEqual(state.executeRequest(self.request(on: eventLoop)).connection, .none) + } + + func testLimitIsSharedAcrossPreferredEventLoopsAndRequiredLoopsCanOverflow() throws { + let loops = (0..<3).map { _ in EmbeddedEventLoop() } + var state = self.makeState() + let first = try self.createdConnection(state.executeRequest(self.request(on: loops[0]))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + let second = try self.createdConnection(state.executeRequest(self.request(on: loops[1]))) + _ = state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1) + XCTAssertEqual(state.executeRequest(self.request(on: loops[2])).connection, .none) + + let required = self.request(on: loops[2], required: true) + let overflow = try self.createdConnection(state.executeRequest(required)) + XCTAssertTrue(overflow.eventLoop === loops[2]) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(overflow, maxConcurrentStreams: 1).request, + .executeRequestsAndCancelTimeouts([required], overflow) + ) + XCTAssertEqual(state.executeRequest(self.request(on: loops[2], required: true)).connection, .none) + } + + func testRequiredLoopCanExpandWithinLimitWithoutUsingOtherLoopsStreams() throws { + let firstLoop = EmbeddedEventLoop() + let requiredLoop = EmbeddedEventLoop() + var state = self.makeState(limit: 3) + let first = try self.createdConnection(state.executeRequest(self.request(on: firstLoop))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 100) + let second = try self.createdConnection(state.executeRequest(self.request(on: requiredLoop, required: true))) + _ = state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1) + let third = try self.createdConnection(state.executeRequest(self.request(on: requiredLoop, required: true))) + XCTAssertTrue(third.eventLoop === requiredLoop) + XCTAssertEqual(state.executeRequest(self.request(on: requiredLoop, required: true)).connection, .none) + } + + func testFailedExpansionRetriesOnlyWhileDemandRemains() throws { + for cancelRequest in [false, true] { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState() + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + let pending = self.request(on: eventLoop) + let second = try self.createdConnection(state.executeRequest(pending)) + guard + case .scheduleBackoffTimer = state.failedToCreateNewConnection( + HTTPClientError.connectTimeout, + connectionID: second.id + ).connection + else { + return XCTFail("Expected a backoff timer") + } + if cancelRequest { + _ = state.cancelRequest(pending.id) + XCTAssertEqual(state.connectionCreationBackoffDone(second.id), .none) + } else { + let retry = try self.createdConnection(state.connectionCreationBackoffDone(second.id)) + XCTAssertNotEqual(retry.id, second.id) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(retry, maxConcurrentStreams: 1).request, + .executeRequestsAndCancelTimeouts([pending], retry) + ) + } + } + } + + func testZeroStreamsCanExpandAndLaterSettingsDrainQueuedRequests() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState() + let pending = self.request(on: eventLoop) + let first = try self.createdConnection(state.executeRequest(pending)) + let zeroStreams = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 0) + guard + case .scheduleTimeoutTimerAndCreateConnection(let timeoutID, let newID, let loop) = + zeroStreams.connection + else { + return XCTFail("Expected idle timeout and another connection, got \(zeroStreams.connection)") + } + XCTAssertEqual(timeoutID, first.id) + XCTAssertEqual(zeroStreams.request, .none) + XCTAssertEqual( + state.newHTTP2MaxConcurrentStreamsReceived(first.id, newMaxStreams: 1).request, + .executeRequestsAndCancelTimeouts([pending], first) + ) + let second = Connection.__testOnly_connection(id: newID, eventLoop: loop) + let lateEstablishment = state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1) + XCTAssertEqual(lateEstablishment.request, .none) + XCTAssertEqual(lateEstablishment.connection, .scheduleTimeoutTimer(second.id, on: loop)) + XCTAssertEqual( + state.connectionIdleTimeout(second.id, on: loop).connection, + .closeConnection(second, isShutdown: .no) + ) + } + + func testGoAwayReplacesCapacityWithoutDuplicateCreation() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState() + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + let second = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1) + let pending = self.request(on: eventLoop) + XCTAssertEqual(state.executeRequest(pending).connection, .none) + let replacement = try self.createdConnection(state.http2ConnectionGoAwayReceived(first.id)) + XCTAssertEqual(state.http2ConnectionGoAwayReceived(first.id), .none) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(replacement, maxConcurrentStreams: 1).request, + .executeRequestsAndCancelTimeouts([pending], replacement) + ) + } + + func testShutdownWaitsForActiveAndStartingConnectionsWithoutExpanding() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState(limit: 3) + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + let pending = self.request(on: eventLoop) + let second = try self.createdConnection(state.executeRequest(pending)) + XCTAssertEqual( + state.shutdown().request, + .failRequestsAndCancelTimeouts([pending], HTTPClientError.cancelled) + ) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1).connection, + .closeConnection(second, isShutdown: .no) + ) + XCTAssertEqual( + state.http2ConnectionStreamClosed(first.id).connection, + .closeConnection(first, isShutdown: .yes(unclean: true)) + ) + } + + func testHTTP1MigrationCanKeepMoreThanOneHTTP2ConnectionUpToLimit() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState(preferHTTP1: true) + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + let second = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + let third = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + let secondEstablished = state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1) + XCTAssertEqual(secondEstablished.connection, .none) + guard case .executeRequestsAndCancelTimeouts = secondEstablished.request else { + return XCTFail("Expected requests on the second connection") + } + XCTAssertEqual( + state.newHTTP2ConnectionCreated(third, maxConcurrentStreams: 1).connection, + .closeConnection(third, isShutdown: .no) + ) + } + + func testHTTP1MigrationPreservesExpansionForQueuedBurst() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState(preferHTTP1: true, http1Limit: 1) + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + let pending = self.request(on: eventLoop) + XCTAssertEqual(state.executeRequest(pending).connection, .none) + let migration = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + guard case .migration(let created, let closed, let timeout) = migration.connection else { + return XCTFail("Expected migration, got \(migration.connection)") + } + XCTAssertEqual(created.count, 1) + XCTAssertTrue(closed.isEmpty) + XCTAssertNil(timeout) + let creation = try XCTUnwrap(created.first) + let second = Connection.__testOnly_connection(id: creation.0, eventLoop: creation.1) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1).request, + .executeRequestsAndCancelTimeouts([pending], second) + ) + } + + func testDefaultLimitRetriesWithoutWaitingForOtherMigrationBackoffs() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState(limit: 1, preferHTTP1: true) + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + let second = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + let third = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.failedToCreateNewConnection(HTTPClientError.connectTimeout, connectionID: second.id) + _ = state.failedToCreateNewConnection(HTTPClientError.connectTimeout, connectionID: third.id) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + + // A backing-off attempt cannot serve the queued requests. Losing the only active + // connection should therefore start a replacement immediately, as with the default pool. + let replacement = try self.createdConnection(state.http2ConnectionClosed(first.id)) + XCTAssertNotEqual(replacement.id, first.id) + // Other timers must not create duplicate attempts while that replacement is starting. + XCTAssertEqual(state.connectionCreationBackoffDone(second.id), .none) + XCTAssertEqual(state.connectionCreationBackoffDone(third.id), .none) + } + + func testSettingsDecreaseDoesNotInvalidateLeasedStreamsOrExceedLimit() throws { + let eventLoop = EmbeddedEventLoop() + var state = self.makeState() + let first = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 2) + _ = state.executeRequest(self.request(on: eventLoop)) + XCTAssertEqual(state.newHTTP2MaxConcurrentStreamsReceived(first.id, newMaxStreams: 0), .none) + let second = try self.createdConnection(state.executeRequest(self.request(on: eventLoop))) + _ = state.newHTTP2ConnectionCreated(second, maxConcurrentStreams: 1) + let pending = self.request(on: eventLoop) + XCTAssertEqual(state.executeRequest(pending).connection, .none) + XCTAssertEqual(state.http2ConnectionStreamClosed(first.id), .none) + XCTAssertEqual( + state.newHTTP2MaxConcurrentStreamsReceived(first.id, newMaxStreams: 2).request, + .executeRequestsAndCancelTimeouts([pending], first) + ) + } + + func testDefaultLimitPreservesSingleConnectionAndRequiredLoopException() throws { + XCTAssertEqual(HTTPClient.Configuration.ConnectionPool().concurrentHTTP2ConnectionsPerHostSoftLimit, 1) + let firstLoop = EmbeddedEventLoop() + let otherLoop = EmbeddedEventLoop() + var state = self.makeState(limit: 1) + let first = try self.createdConnection(state.executeRequest(self.request(on: firstLoop))) + _ = state.newHTTP2ConnectionCreated(first, maxConcurrentStreams: 1) + XCTAssertEqual(state.executeRequest(self.request(on: otherLoop)).connection, .none) + let required = self.request(on: otherLoop, required: true) + let overflow = try self.createdConnection(state.executeRequest(required)) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(overflow, maxConcurrentStreams: 1).request, + .executeRequestsAndCancelTimeouts([required], overflow) + ) + } + + func testIdleTimeoutExpandsForQueuedRequestsOnAnotherRequiredEventLoop() throws { + let idleLoop = EmbeddedEventLoop() + let busyLoop = EmbeddedEventLoop() + var state = self.makeState() + let idle = try self.createdConnection(state.executeRequest(self.request(on: idleLoop))) + _ = state.newHTTP2ConnectionCreated(idle, maxConcurrentStreams: 1) + _ = state.http2ConnectionStreamClosed(idle.id) + let busy = try self.createdConnection(state.executeRequest(self.request(on: busyLoop, required: true))) + _ = state.newHTTP2ConnectionCreated(busy, maxConcurrentStreams: 1) + let pending = self.request(on: busyLoop, required: true) + XCTAssertEqual(state.executeRequest(pending).connection, .none) + + let timeout = state.connectionIdleTimeout(idle.id, on: idleLoop) + guard case .closeConnectionAndCreateConnection(let closed, let newID, let loop) = timeout.connection else { + return XCTFail("Expected replacement on the busy event loop, got \(timeout.connection)") + } + XCTAssertEqual(closed, idle) + XCTAssertTrue(loop === busyLoop) + let additional = Connection.__testOnly_connection(id: newID, eventLoop: loop) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(additional, maxConcurrentStreams: 1).request, + .executeRequestsAndCancelTimeouts([pending], additional) + ) + } + + func testUnexpectedCloseExpandsForQueuedRequestsOnAnotherRequiredEventLoop() throws { + let closingLoop = EmbeddedEventLoop() + let busyLoop = EmbeddedEventLoop() + var state = self.makeState() + let closing = try self.createdConnection(state.executeRequest(self.request(on: closingLoop))) + _ = state.newHTTP2ConnectionCreated(closing, maxConcurrentStreams: 1) + let busy = try self.createdConnection(state.executeRequest(self.request(on: busyLoop, required: true))) + _ = state.newHTTP2ConnectionCreated(busy, maxConcurrentStreams: 1) + let pending = self.request(on: busyLoop, required: true) + XCTAssertEqual(state.executeRequest(pending).connection, .none) + + let additional = try self.createdConnection(state.http2ConnectionClosed(closing.id)) + XCTAssertTrue(additional.eventLoop === busyLoop) + XCTAssertEqual( + state.newHTTP2ConnectionCreated(additional, maxConcurrentStreams: 1).request, + .executeRequestsAndCancelTimeouts([pending], additional) + ) + } +} diff --git a/Tests/AsyncHTTPClientTests/SwiftConfigurationTests.swift b/Tests/AsyncHTTPClientTests/SwiftConfigurationTests.swift index e80d49ed8..cd2f9ff95 100644 --- a/Tests/AsyncHTTPClientTests/SwiftConfigurationTests.swift +++ b/Tests/AsyncHTTPClientTests/SwiftConfigurationTests.swift @@ -39,6 +39,7 @@ struct HTTPClientConfigurationPropsTests { "connectionPool.idleTimeoutMs": 120_000, "connectionPool.concurrentHTTP1ConnectionsPerHostSoftLimit": 16, + "connectionPool.concurrentHTTP2ConnectionsPerHostSoftLimit": 3, "connectionPool.retryConnectionEstablishment": false, "connectionPool.preWarmedHTTP1ConnectionCount": 5, @@ -77,6 +78,7 @@ struct HTTPClientConfigurationPropsTests { #expect(config.connectionPool.idleTimeout == .milliseconds(120000)) #expect(config.connectionPool.concurrentHTTP1ConnectionsPerHostSoftLimit == 16) + #expect(config.connectionPool.concurrentHTTP2ConnectionsPerHostSoftLimit == 3) #expect(config.connectionPool.retryConnectionEstablishment == false) #expect(config.connectionPool.preWarmedHTTP1ConnectionCount == 5) @@ -110,6 +112,7 @@ struct HTTPClientConfigurationPropsTests { #expect(config.connectionPool.idleTimeout == .seconds(60)) #expect(config.connectionPool.concurrentHTTP1ConnectionsPerHostSoftLimit == 8) + #expect(config.connectionPool.concurrentHTTP2ConnectionsPerHostSoftLimit == 1) #expect(config.connectionPool.retryConnectionEstablishment == true) #expect(config.connectionPool.preWarmedHTTP1ConnectionCount == 0)