Files
async-http-client/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1Connections.swift
T
George BarnettandGitHub 6c5058ee2c Add a control to limit connection reuses (#678)
Motivation:

Sometimes it can be helpful to limit the number of times a connection
can be used before discarding it. AHC has no such support for this at
the moment.

Modifications:

- Add a `maximumUsesPerConnection` configuration option which defaults
  to `nil` (i.e. no limit).
- For HTTP1 we count down uses in the state machine and close the
  connection if it hits zero.
- For HTTP2, each use maps to a stream so we count down remaining uses
  in the state machine which we combine with max concurrent streams to
  limit how many streams are available per connection. We also count
  remaining uses in the HTTP2 idle handler: we treat no remaining uses
  as receiving a GOAWAY frame and notify the pool which then drains the
  streams and replaces the connection.

Result:

Users can control how many times each connection can be used.
2023-04-11 14:49:44 +01:00

830 lines
36 KiB
Swift

//===----------------------------------------------------------------------===//
//
// This source file is part of the AsyncHTTPClient open source project
//
// Copyright (c) 2021 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
extension HTTPConnectionPool {
/// Represents the state of a single HTTP/1.1 connection
private struct HTTP1ConnectionState {
enum State {
/// the connection is creating a connection. Valid transitions are to: .backingOff, .idle, and .closed
case starting(maximumUses: Int?)
/// the connection is waiting to retry the establishing a connection. Valid transitions are to: .closed.
/// This means, the connection can be removed from the connections without cancelling external
/// state. The connection state can then be replaced by a new one.
case backingOff
/// the connection is idle for a new request. Valid transitions to: .leased and .closed
case idle(Connection, since: NIODeadline, remainingUses: Int?)
/// the connection is leased and running for a request. Valid transitions to: .idle and .closed
case leased(Connection, remainingUses: Int?)
/// the connection is closed. final state.
case closed
}
private var state: State
let connectionID: Connection.ID
let eventLoop: EventLoop
init(connectionID: Connection.ID, eventLoop: EventLoop, maximumUses: Int?) {
self.connectionID = connectionID
self.eventLoop = eventLoop
self.state = .starting(maximumUses: maximumUses)
}
var isConnecting: Bool {
switch self.state {
case .starting:
return true
case .backingOff, .closed, .idle, .leased:
return false
}
}
var isBackingOff: Bool {
switch self.state {
case .backingOff:
return true
case .starting, .closed, .idle, .leased:
return false
}
}
var isIdle: Bool {
switch self.state {
case .idle:
return true
case .backingOff, .starting, .leased, .closed:
return false
}
}
var idleAndNoRemainingUses: Bool {
switch self.state {
case .idle(_, since: _, remainingUses: let remainingUses):
if let remainingUses = remainingUses {
return remainingUses <= 0
} else {
return false
}
case .backingOff, .starting, .leased, .closed:
return false
}
}
var canOrWillBeAbleToExecuteRequests: Bool {
switch self.state {
case .leased, .backingOff, .idle, .starting:
return true
case .closed:
return false
}
}
var isLeased: Bool {
switch self.state {
case .leased:
return true
case .backingOff, .starting, .idle, .closed:
return false
}
}
var idleSince: NIODeadline? {
switch self.state {
case .idle(_, since: let idleSince, _):
return idleSince
case .backingOff, .starting, .leased, .closed:
return nil
}
}
var isClosed: Bool {
switch self.state {
case .closed:
return true
case .idle, .starting, .leased, .backingOff:
return false
}
}
mutating func connected(_ connection: Connection) {
switch self.state {
case .starting(maximumUses: let maxUses):
self.state = .idle(connection, since: .now(), remainingUses: maxUses)
case .backingOff, .idle, .leased, .closed:
preconditionFailure("Invalid state: \(self.state)")
}
}
/// The connection failed to start
mutating func failedToConnect() {
switch self.state {
case .starting:
self.state = .backingOff
case .backingOff, .idle, .leased, .closed:
preconditionFailure("Invalid state: \(self.state)")
}
}
mutating func lease() -> Connection {
switch self.state {
case .idle(let connection, since: _, remainingUses: let remainingUses):
self.state = .leased(connection, remainingUses: remainingUses.map { $0 - 1 })
return connection
case .backingOff, .starting, .leased, .closed:
preconditionFailure("Invalid state: \(self.state)")
}
}
mutating func release() {
switch self.state {
case .leased(let connection, let remainingUses):
self.state = .idle(connection, since: .now(), remainingUses: remainingUses)
case .backingOff, .starting, .idle, .closed:
preconditionFailure("Invalid state: \(self.state)")
}
}
mutating func close() -> Connection {
switch self.state {
case .idle(let connection, since: _, remainingUses: _):
self.state = .closed
return connection
case .backingOff, .starting, .leased, .closed:
preconditionFailure("Invalid state: \(self.state)")
}
}
mutating func fail() {
switch self.state {
case .starting, .backingOff, .idle, .leased:
self.state = .closed
case .closed:
preconditionFailure("Invalid state: \(self.state)")
}
}
enum CleanupAction {
case removeConnection
case keepConnection
}
/// Cleanup the current connection for shutdown.
///
/// This method is called, when the connections shall shutdown. Depending on the state
/// the connection is in, it adds itself to one of the arrays that are used to signal shutdown
/// intent to the underlying connections. Connections that are backing off can be easily
/// dropped (since, we only need to cancel the backoff timer), connections that are leased
/// need to be cancelled (notifying the `ChannelHandler` that we want to cancel the
/// running request), connections that are idle can be closed right away. Sadly we can't
/// cancel connection starts right now. For this reason we need to wait for them to succeed
/// or fail until we finalize the shutdown.
///
/// - Parameter context: A cleanup context to add the connection to based on its state.
/// - Returns: A cleanup action indicating if the connection can be removed from the
/// connection list.
func cleanup(_ context: inout CleanupContext) -> CleanupAction {
switch self.state {
case .backingOff:
context.connectBackoff.append(self.connectionID)
return .removeConnection
case .starting:
return .keepConnection
case .idle(let connection, since: _, remainingUses: _):
context.close.append(connection)
return .removeConnection
case .leased(let connection, remainingUses: _):
context.cancel.append(connection)
return .keepConnection
case .closed:
preconditionFailure("Unexpected state: Did not expect to have connections with this state in the state machine: \(self.state)")
}
}
enum MigrateAction {
case removeConnection
case keepConnection
}
func migrateToHTTP2(_ context: inout HTTP1Connections.HTTP1ToHTTP2MigrationContext) -> MigrateAction {
switch self.state {
case .starting:
context.starting.append((self.connectionID, self.eventLoop))
return .removeConnection
case .backingOff:
context.backingOff.append((self.connectionID, self.eventLoop))
return .removeConnection
case .idle(let connection, since: _, remainingUses: _):
// Idle connections can be removed right away
context.close.append(connection)
return .removeConnection
case .leased:
return .keepConnection
case .closed:
preconditionFailure("Unexpected state: Did not expect to have connections with this state in the state machine: \(self.state)")
}
}
}
/// A structure to hold the currently active HTTP/1.1 connections.
///
/// The general purpose connection pool (pool for requests that don't have special `EventLoop`
/// requirements) will grow up until `maximumConcurrentConnections`. If requests have
/// special `EventLoop` requirements overflow connections might be opened.
///
/// All connections live in the same `connections` array. In the front are the general purpose
/// connections. In the back (starting with the `overflowIndex`) are the connections for
/// requests with special needs.
struct HTTP1Connections {
/// The maximum number of connections in the general purpose pool.
private let maximumConcurrentConnections: Int
/// A connectionID generator.
private let generator: Connection.ID.Generator
/// The connections states
private var connections: [HTTP1ConnectionState]
/// The index after which you will find the connections for requests with `EventLoop`
/// requirements in `connections`.
private var overflowIndex: Array<HTTP1ConnectionState>.Index
/// The number of times each connection can be used before it is closed and replaced.
private let maximumConnectionUses: Int?
init(maximumConcurrentConnections: Int, generator: Connection.ID.Generator, maximumConnectionUses: Int?) {
self.connections = []
self.connections.reserveCapacity(maximumConcurrentConnections)
self.overflowIndex = self.connections.endIndex
self.maximumConcurrentConnections = maximumConcurrentConnections
self.generator = generator
self.maximumConnectionUses = maximumConnectionUses
}
var stats: Stats {
var stats = Stats()
// all additions here can be unchecked, since we will have at max self.connections.count
// which itself is an Int. For this reason we will never overflow.
for connectionState in self.connections {
if connectionState.isConnecting {
stats.connecting &+= 1
} else if connectionState.isBackingOff {
stats.backingOff &+= 1
} else if connectionState.isLeased {
stats.leased &+= 1
} else if connectionState.isIdle {
stats.idle &+= 1
}
}
return stats
}
var isEmpty: Bool {
self.connections.isEmpty
}
var canGrow: Bool {
self.overflowIndex < self.maximumConcurrentConnections
}
var startingGeneralPurposeConnections: Int {
var connecting = 0
for connectionState in self.connections[0..<self.overflowIndex] {
if connectionState.isConnecting || connectionState.isBackingOff {
// connecting can't be greater than self.connections.count so it can't overflow.
// For this reason we can save the bounds check.
connecting &+= 1
}
}
return connecting
}
private var maximumAdditionalGeneralPurposeConnections: Int {
self.maximumConcurrentConnections - (self.overflowIndex - 1)
}
/// Is there at least one connection that is able to run requests
var hasActiveConnections: Bool {
self.connections.contains(where: { $0.isIdle || $0.isLeased })
}
func startingEventLoopConnections(on eventLoop: EventLoop) -> Int {
return self.connections[self.overflowIndex..<self.connections.endIndex].reduce(into: 0) { count, connection in
guard connection.eventLoop === eventLoop else { return }
if connection.isConnecting || connection.isBackingOff {
count &+= 1
}
}
}
// MARK: - Mutations -
/// A connection's use. Did it serve in the pool or was it specialized for an `EventLoop`?
enum ConnectionUse {
case generalPurpose
case eventLoop(EventLoop)
}
/// Information around an idle connection.
struct IdleConnectionContext {
/// The `EventLoop` the connection runs on.
var eventLoop: EventLoop
/// The connection's use. Either general purpose or for requests with `EventLoop`
/// requirements.
var use: ConnectionUse
/// Whether the connection should be closed.
var shouldBeClosed: Bool
}
/// Information around the failed/closed connection.
struct FailedConnectionContext {
/// The eventLoop the connection ran on.
var eventLoop: EventLoop
/// The failed connection's use.
var use: ConnectionUse
/// Connections that we start up for this use-case
var connectionsStartingForUseCase: Int
}
struct HTTP1ToHTTP2MigrationContext {
var backingOff: [(Connection.ID, EventLoop)] = []
var starting: [(Connection.ID, EventLoop)] = []
var close: [Connection] = []
}
// MARK: Connection creation
mutating func createNewConnection(on eventLoop: EventLoop) -> Connection.ID {
precondition(self.canGrow)
let connection = HTTP1ConnectionState(
connectionID: self.generator.next(),
eventLoop: eventLoop,
maximumUses: self.maximumConnectionUses
)
self.connections.insert(connection, at: self.overflowIndex)
self.overflowIndex = self.connections.index(after: self.overflowIndex)
return connection.connectionID
}
mutating func createNewOverflowConnection(on eventLoop: EventLoop) -> Connection.ID {
let connection = HTTP1ConnectionState(
connectionID: self.generator.next(),
eventLoop: eventLoop,
maximumUses: self.maximumConnectionUses
)
self.connections.append(connection)
return connection.connectionID
}
/// A new HTTP/1.1 connection was established.
///
/// This will put the connection into the idle state.
///
/// - Parameter connection: The new established connection.
/// - Returns: An index and an IdleConnectionContext to determine the next action for the now idle connection.
/// Call ``leaseConnection(at:)`` or ``closeConnection(at:)`` with the supplied index after
/// this.
mutating func newHTTP1ConnectionEstablished(_ connection: Connection) -> (Int, IdleConnectionContext) {
guard let index = self.connections.firstIndex(where: { $0.connectionID == connection.id }) else {
preconditionFailure("There is a new connection that we didn't request!")
}
precondition(connection.eventLoop === self.connections[index].eventLoop, "Expected the new connection to be on EL")
self.connections[index].connected(connection)
let context = self.generateIdleConnectionContextForConnection(at: index)
return (index, context)
}
/// Move the HTTP1ConnectionState to backingOff.
///
/// - Parameter connectionID: The connectionID of the failed connection attempt
/// - Returns: The eventLoop on which to schedule the backoff timer
mutating func backoffNextConnectionAttempt(_ connectionID: Connection.ID) -> EventLoop {
guard let index = self.connections.firstIndex(where: { $0.connectionID == connectionID }) else {
preconditionFailure("We tried to create a new connection that we know nothing about?")
}
self.connections[index].failedToConnect()
return self.connections[index].eventLoop
}
// MARK: Leasing and releasing
/// Lease a connection on the preferred EventLoop
///
/// If no connection is available on the preferred EventLoop, a connection on
/// another eventLoop might be returned, if one is available.
///
/// - Parameter eventLoop: The preferred EventLoop for the request
/// - Returns: A connection to execute a request on.
mutating func leaseConnection(onPreferred eventLoop: EventLoop) -> Connection? {
guard let index = self.findIdleConnection(onPreferred: eventLoop) else {
return nil
}
return self.connections[index].lease()
}
/// Lease a connection on the required EventLoop
///
/// If no connection is available on the required EventLoop nil is returned.
///
/// - Parameter eventLoop: The required EventLoop for the request
/// - Returns: A connection to execute a request on.
mutating func leaseConnection(onRequired eventLoop: EventLoop) -> Connection? {
guard let index = self.findIdleConnection(onRequired: eventLoop) else {
return nil
}
return self.connections[index].lease()
}
mutating func leaseConnection(at index: Int) -> Connection {
self.connections[index].lease()
}
func parkConnection(at index: Int) -> (Connection.ID, EventLoop) {
precondition(self.connections[index].isIdle)
return (self.connections[index].connectionID, self.connections[index].eventLoop)
}
/// A new HTTP/1.1 connection was released.
///
/// This will put the position into the idle state.
///
/// - Parameter connectionID: The released connection's id.
/// - Returns: An index and an IdleConnectionContext to determine the next action for the now idle connection.
/// Call ``leaseConnection(at:)`` or ``closeConnection(at:)`` with the supplied index after
/// this. If you want to park the connection no further call is required.
mutating func releaseConnection(_ connectionID: Connection.ID) -> (Int, IdleConnectionContext) {
guard let index = self.connections.firstIndex(where: { $0.connectionID == connectionID }) else {
preconditionFailure("A connection that we don't know was released? Something is very wrong...")
}
self.connections[index].release()
let context = self.generateIdleConnectionContextForConnection(at: index)
return (index, context)
}
// MARK: Connection close/removal
/// Closes the connection at the given index. This will also remove the connection right away.
mutating func closeConnection(at index: Int) -> Connection {
if index < self.overflowIndex {
self.overflowIndex = self.connections.index(before: self.overflowIndex)
}
var connectionState = self.connections.remove(at: index)
return connectionState.close()
}
mutating func removeConnection(at index: Int) {
precondition(self.connections[index].isClosed)
if index < self.overflowIndex {
self.overflowIndex = self.connections.index(before: self.overflowIndex)
}
self.connections.remove(at: index)
}
mutating func closeConnectionIfIdle(_ connectionID: Connection.ID) -> Connection? {
guard let index = self.connections.firstIndex(where: { $0.connectionID == connectionID }) else {
// because of a race this connection (connection close runs against trigger of timeout)
// was already removed from the state machine.
return nil
}
guard self.connections[index].isIdle else {
// connection is not idle anymore, we may have just leased it for a request
return nil
}
return self.closeConnection(at: index)
}
mutating func replaceConnection(at index: Int) -> (Connection.ID, EventLoop) {
precondition(self.connections[index].isClosed)
let newConnection = HTTP1ConnectionState(
connectionID: self.generator.next(),
eventLoop: self.connections[index].eventLoop,
maximumUses: self.maximumConnectionUses
)
self.connections[index] = newConnection
return (newConnection.connectionID, newConnection.eventLoop)
}
// MARK: Connection failure
/// Fail a connection. Call this method, if a connection suddenly closed, did not startup correctly,
/// or the backoff time is done.
///
/// This will put the position into the closed state.
///
/// - Parameter connectionID: The failed connection's id.
/// - Returns: An optional index and an IdleConnectionContext to determine the next action for the closed connection.
/// You must call ``removeConnection(at:)`` or ``replaceConnection(at:)`` with the
/// supplied index after this. If nil is returned the connection was closed by the state machine and was
/// therefore already removed.
mutating func failConnection(_ connectionID: Connection.ID) -> (Int, FailedConnectionContext)? {
guard let index = self.connections.firstIndex(where: { $0.connectionID == connectionID }) else {
return nil
}
let use: ConnectionUse
self.connections[index].fail()
let eventLoop = self.connections[index].eventLoop
let starting: Int
if index < self.overflowIndex {
use = .generalPurpose
starting = self.startingGeneralPurposeConnections
} else {
use = .eventLoop(eventLoop)
starting = self.startingEventLoopConnections(on: eventLoop)
}
let context = FailedConnectionContext(
eventLoop: eventLoop,
use: use,
connectionsStartingForUseCase: starting
)
return (index, context)
}
// MARK: Migration
mutating func migrateToHTTP2() -> HTTP1ToHTTP2MigrationContext {
var migrationContext = HTTP1ToHTTP2MigrationContext()
let initialOverflowIndex = self.overflowIndex
self.connections = self.connections.enumerated().compactMap { index, connectionState in
switch connectionState.migrateToHTTP2(&migrationContext) {
case .removeConnection:
// If the connection has an index smaller than the previous overflow index,
// we deal with a general purpose connection.
// For this reason we need to decrement the overflow index.
if index < initialOverflowIndex {
self.overflowIndex = self.connections.index(before: self.overflowIndex)
}
return nil
case .keepConnection:
return connectionState
}
}
return migrationContext
}
/// We only handle starting and backing off connection here.
/// All already running connections must be handled by the enclosing state machine.
/// - Parameters:
/// - starting: starting HTTP connections from previous state machine
/// - backingOff: backing off HTTP connections from previous state machine
mutating func migrateFromHTTP2(
starting: [(Connection.ID, EventLoop)],
backingOff: [(Connection.ID, EventLoop)]
) {
for (connectionID, eventLoop) in starting {
let newConnection = HTTP1ConnectionState(
connectionID: connectionID,
eventLoop: eventLoop,
maximumUses: self.maximumConnectionUses
)
self.connections.insert(newConnection, at: self.overflowIndex)
/// If we can grow, we mark the connection as a general purpose connection.
/// Otherwise, it will be an overflow connection which is only used once for requests with a required event loop
if self.canGrow {
self.overflowIndex = self.connections.index(after: self.overflowIndex)
}
}
for (connectionID, eventLoop) in backingOff {
var backingOffConnection = HTTP1ConnectionState(
connectionID: connectionID,
eventLoop: eventLoop,
maximumUses: self.maximumConnectionUses
)
// TODO: Maybe we want to add a static init for backing off connections to HTTP1ConnectionState
backingOffConnection.failedToConnect()
self.connections.insert(backingOffConnection, at: self.overflowIndex)
/// If we can grow, we mark the connection as a general purpose connection.
/// Otherwise, it will be an overflow connection which is only used once for requests with a required event loop
if self.canGrow {
self.overflowIndex = self.connections.index(after: self.overflowIndex)
}
}
}
/// We will create new connections for each `requiredEventLoopOfPendingRequests`
/// In addition, we also create more general purpose connections if we do not have enough to execute
/// all requests on the given `preferredEventLoopsOfPendingGeneralPurposeRequests`
/// until we reach `maximumConcurrentConnections`
/// - Parameters:
/// - requiredEventLoopsForPendingRequests:
/// event loops for which we have requests with a required event loop.
/// Duplicates are not allowed.
/// - generalPurposeRequestCountPerPreferredEventLoop:
/// request count with no required event loop,
/// grouped by preferred event loop and ordered descending by number of requests
/// - Returns: new connections that must be created
mutating func createConnectionsAfterMigrationIfNeeded(
requiredEventLoopOfPendingRequests: [(EventLoop, Int)],
generalPurposeRequestCountGroupedByPreferredEventLoop: [(EventLoop, Int)]
) -> [(Connection.ID, EventLoop)] {
// create new connections for requests with a required event loop
// we may already start connections for those requests and do not want to start to many
let startingRequiredEventLoopConnectionCount = Dictionary(
self.connections[self.overflowIndex..<self.connections.endIndex].lazy.map {
($0.eventLoop.id, 1)
},
uniquingKeysWith: +
)
var connectionToCreate = requiredEventLoopOfPendingRequests
.flatMap { eventLoop, requestCount -> [(Connection.ID, EventLoop)] in
// We need a connection for each queued request with a required event loop.
// Therefore, we look how many request we have queued for a given `eventLoop` and
// how many connections we are already starting on the given `eventLoop`.
// If we have not enough, we will create additional connections to have at least
// on connection per request.
let connectionsToStart = requestCount - startingRequiredEventLoopConnectionCount[eventLoop.id, default: 0]
return stride(from: 0, to: connectionsToStart, by: 1).lazy.map { _ in
(self.createNewOverflowConnection(on: eventLoop), eventLoop)
}
}
// create new connections for requests without a required event loop
// TODO: improve algorithm to create connections uniformly across all preferred event loops
// while paying attention to the number of queued request per event loop
// Currently we start by creating new connections on the event loop with the most queued
// requests. If we have created enough connections to cover all requests for the first
// event loop we will continue with the event loop with the second most queued requests
// and so on and so forth. The `generalPurposeRequestCountGroupedByPreferredEventLoop`
// array is already ordered so we can just iterate over it without sorting by request count.
let newGeneralPurposeConnections: [(Connection.ID, EventLoop)] = generalPurposeRequestCountGroupedByPreferredEventLoop
// we do not want to allocated intermediate arrays.
.lazy
// we flatten the grouped list of event loops by lazily repeating the event loop
// for each request.
// As a result we get one event loop per request (`[EventLoop]`).
.flatMap { eventLoop, requestCount in
repeatElement(eventLoop, count: requestCount)
}
// we may already start connections and do not want to start too many
.dropLast(self.startingGeneralPurposeConnections)
// we need to respect the used defined `maximumConcurrentConnections`
.prefix(self.maximumAdditionalGeneralPurposeConnections)
// we now create a connection for each remaining event loop
.map { eventLoop in
(self.createNewConnection(on: eventLoop), eventLoop)
}
connectionToCreate.append(contentsOf: newGeneralPurposeConnections)
return connectionToCreate
}
// MARK: Shutdown
mutating func shutdown() -> CleanupContext {
var cleanupContext = CleanupContext()
let initialOverflowIndex = self.overflowIndex
self.connections = self.connections.enumerated().compactMap { index, connectionState in
switch connectionState.cleanup(&cleanupContext) {
case .removeConnection:
// If the connection has an index smaller than the previous overflow index,
// we deal with a general purpose connection.
// For this reason we need to decrement the overflow index.
if index < initialOverflowIndex {
self.overflowIndex = self.connections.index(before: self.overflowIndex)
}
return nil
case .keepConnection:
return connectionState
}
}
return cleanupContext
}
// MARK: - Private functions -
private func generateIdleConnectionContextForConnection(at index: Int) -> IdleConnectionContext {
precondition(self.connections[index].isIdle)
let eventLoop = self.connections[index].eventLoop
let use: ConnectionUse
if index < self.overflowIndex {
use = .generalPurpose
} else {
use = .eventLoop(eventLoop)
}
let hasNoRemainingUses = self.connections[index].idleAndNoRemainingUses
return IdleConnectionContext(eventLoop: eventLoop, use: use, shouldBeClosed: hasNoRemainingUses)
}
private func findIdleConnection(onPreferred preferredEL: EventLoop) -> Int? {
var eventLoopMatch: (Int, NIODeadline)?
var goodMatch: (Int, NIODeadline)?
// To find an appropriate connection we iterate all existing connections.
// While we do this we try to find the best fitting connection for our request.
//
// A perfect match, runs on the same eventLoop and has been idle the shortest amount
// of time (returned the most recently).
//
// An okay match is not on the same eventLoop, and has been idle for the shortest
// time.
for (index, conn) in self.connections.enumerated() {
guard let connReturn = conn.idleSince else {
continue
}
if conn.eventLoop === preferredEL {
switch eventLoopMatch {
case .none:
eventLoopMatch = (index, connReturn)
case .some((_, let existingMatchReturn)) where connReturn > existingMatchReturn:
eventLoopMatch = (index, connReturn)
default:
break
}
} else {
switch goodMatch {
case .none:
goodMatch = (index, connReturn)
case .some((_, let existingMatchReturn)):
// We don't require a specific eventLoop. For this reason we want to pick a
// matching eventLoop that has been idle the shortest.
if connReturn > existingMatchReturn {
goodMatch = (index, connReturn)
}
}
}
}
if let (index, _) = eventLoopMatch {
return index
}
if let (index, _) = goodMatch {
return index
}
return nil
}
func findIdleConnection(onRequired requiredEL: EventLoop) -> Int? {
var match: (Int, NIODeadline)?
// To find an appropriate connection we iterate all existing connections.
// While we do this we try to find the best fitting connection for our request.
//
// A match, runs on the same eventLoop and has been idle the shortest amount of time.
for (index, conn) in self.connections.enumerated() {
// 1. Ensure we are on the correct EL.
guard conn.eventLoop === requiredEL else {
continue
}
// 2. Ensure the connection is idle
guard let connReturn = conn.idleSince else {
continue
}
switch match {
case .none:
match = (index, connReturn)
case .some((_, let existingMatchReturn)) where connReturn > existingMatchReturn:
// the currently iterated eventLoop has been idle for a shorter amount than
// the current best match.
match = (index, connReturn)
default:
// the currently iterated eventLoop has been idle for a longer amount than
// the current best match. We continue the iteration.
continue
}
}
if let (index, _) = match {
return index
}
return nil
}
struct Stats {
var idle: Int = 0
var leased: Int = 0
var connecting: Int = 0
var backingOff: Int = 0
}
}
}