Add support for HTTP/1 connection pre-warming (#856)

Motivation This patch adds support for HTTP/1 connection pre-warming. This allows the user to request that the HTTP/1 connection pool create and maintain extra connections, above-and-beyond those strictly needed to run the pool. This pool can be used to absorb small spikes in request traffic without increasing latency to account for connection creation. Modifications - Added new configuration properties for pre-warmed connections. - Amended the HTTP/1 state machine to create new connections where necessary. - Added state machine tests. Results Pre-warmed connections are available.

Author: Lukasa

Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift

@@ -79,7 +79,8 @@ final class HTTPConnectionPool:
 .concurrentHTTP1ConnectionsPerHostSoftLimit,
  retryConnectionEstablishment: clientConfiguration.connectionPool.retryConnectionEstablishment,
  preferHTTP1: clientConfiguration.httpVersion ==.http1Only,
- maximumConnectionUses: clientConfiguration.maximumUsesPerConnection
+ maximumConnectionUses: clientConfiguration.maximumUsesPerConnection,
+ preWarmedHTTP1ConnectionCount: clientConfiguration.connectionPool.preWarmedHTTP1ConnectionCount
 )
 }
 
@@ -104,6 +105,11 @@ final class HTTPConnectionPool:
 enumUnlocked{
 case createConnection(Connection.ID, on:EventLoop)
 case closeConnection(Connection, isShutdown:StateMachine.ConnectionAction.IsShutdown)
+case closeConnectionAndCreateConnection(
+ close:Connection,
+ newConnectionID:Connection.ID,
+ on:EventLoop
+)
 case cleanupConnections(CleanupContext, isShutdown:StateMachine.ConnectionAction.IsShutdown)
 case migration(
  createConnections:[(Connection.ID,EventLoop)],
@@ -185,12 +191,27 @@ final class HTTPConnectionPool:
 self.locked.connection =.scheduleBackoffTimer(connectionID, backoff: backoff, on: eventLoop)
 case.scheduleTimeoutTimer(let connectionID, on:let eventLoop):
 self.locked.connection =.scheduleTimeoutTimer(connectionID, on: eventLoop)
+case.scheduleTimeoutTimerAndCreateConnection(let timeoutID,let newConnectionID,let eventLoop):
+self.locked.connection =.scheduleTimeoutTimer(timeoutID, on: eventLoop)
+self.unlocked.connection =.createConnection(newConnectionID, on: eventLoop)
 case.cancelTimeoutTimer(let connectionID):
 self.locked.connection =.cancelTimeoutTimer(connectionID)
+case.createConnectionAndCancelTimeoutTimer(let createdID, on:let eventLoop, cancelTimerID:let cancelID):
+self.unlocked.connection =.createConnection(createdID, on: eventLoop)
+self.locked.connection =.cancelTimeoutTimer(cancelID)
 case.closeConnection(let connection,let isShutdown):
 self.unlocked.connection =.closeConnection(connection, isShutdown: isShutdown)
+case.closeConnectionAndCreateConnection(
+let closeConnection,
+let newConnectionID,
+let eventLoop
+):
+self.unlocked.connection =.closeConnectionAndCreateConnection(
+ close: closeConnection,
+ newConnectionID: newConnectionID,
+ on: eventLoop
+)
 case.cleanupConnections(var cleanupContext,let isShutdown):
- //
 self.locked.connection =.cancelBackoffTimers(cleanupContext.connectBackoff)
  cleanupContext.connectBackoff =[]
 self.unlocked.connection =.cleanupConnections(cleanupContext, isShutdown: isShutdown)
@@ -287,6 +308,23 @@ final class HTTPConnectionPool:
 self.delegate.connectionPoolDidShutdown(self, unclean: unclean)
 }
 
+case.closeConnectionAndCreateConnection(
+let connectionToClose,
+let newConnectionID,
+let eventLoop
+):
+self.logger.trace(
+"closing and creating connection",
+ metadata:[
+"ahc-connection-id":"\(connectionToClose.id)"
+]
+)
+
+self.createConnection(newConnectionID, on: eventLoop)
+
+ // we are not interested in the close promise...
+ connectionToClose.close(promise:nil)
+
 case.cleanupConnections(let cleanupContext,let isShutdown):
 forconnectionin cleanupContext.close {
  connection.close(promise:nil)
@@ -400,7 +438,7 @@ final class HTTPConnectionPool:
 self.modifyStateAndRunActions{ stateMachine in
 ifself._idleTimer.removeValue(forKey: connectionID)!=nil{
  // The timer still exists. State Machines assumes it is alive
-return stateMachine.connectionIdleTimeout(connectionID)
+return stateMachine.connectionIdleTimeout(connectionID, on: eventLoop)
 }
 return.none
 }

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1Connections.swift

@@ -262,32 +262,30 @@ extension HTTPConnectionPool{
 privatevaroverflowIndex:Array<HTTP1ConnectionState>.Index
  /// The number of times each connection can be used before it is closed and replaced.
 privateletmaximumConnectionUses:Int?
-
-init(maximumConcurrentConnections:Int, generator:Connection.ID.Generator, maximumConnectionUses:Int?){
+ /// How many pre-warmed connections we should create.
+privateletpreWarmedConnectionCount:Int
+
+init(
+ maximumConcurrentConnections:Int,
+ generator:Connection.ID.Generator,
+ maximumConnectionUses:Int?,
+ preWarmedHTTP1ConnectionCount:Int
+){
 self.connections =[]
 self.connections.reserveCapacity(min(maximumConcurrentConnections,1024))
 self.overflowIndex =self.connections.endIndex
 self.maximumConcurrentConnections = maximumConcurrentConnections
 self.generator = generator
 self.maximumConnectionUses = maximumConnectionUses
+self.preWarmedConnectionCount = preWarmedHTTP1ConnectionCount
 }
 
 varstats:Stats{
-varstats=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.
-forconnectionStateinself.connections {
-if connectionState.isConnecting {
- stats.connecting &+=1
-}elseif connectionState.isBackingOff {
- stats.backingOff &+=1
-}elseif connectionState.isLeased {
- stats.leased &+=1
-}elseif connectionState.isIdle {
- stats.idle &+=1
-}
-}
-return stats
+self.connectionStats(in:self.connections.startIndex..<self.connections.endIndex)
+}
+
+vargeneralPurposeStats:Stats{
+self.connectionStats(in:self.connections.startIndex..<self.overflowIndex)
 }
 
 varisEmpty:Bool{
@@ -328,6 +326,24 @@ extension HTTPConnectionPool{
 }
 }
 
+privatefunc connectionStats(in range:Range<Int>)->Stats{
+varstats=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.
+forconnectionStateinself.connections[range]{
+if connectionState.isConnecting {
+ stats.connecting &+=1
+}elseif connectionState.isBackingOff {
+ stats.backingOff &+=1
+}elseif connectionState.isLeased {
+ stats.leased &+=1
+}elseif connectionState.isIdle {
+ stats.idle &+=1
+}
+}
+return stats
+}
+
  // MARK: - Mutations -
 
  /// A connection's use. Did it serve in the pool or was it specialized for an `EventLoop`?
@@ -836,6 +852,10 @@ extension HTTPConnectionPool{
 varleased:Int=0
 varconnecting:Int=0
 varbackingOff:Int=0
+
+varnonLeased:Int{
+self.idle +self.connecting +self.backingOff
+}
 }
 }
 }

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift

@@ -33,19 +33,23 @@ extension HTTPConnectionPool{
  /// The property was introduced to fail fast during testing.
  /// Otherwise this should always be true and not turned off.
 privateletretryConnectionEstablishment:Bool
+privateletpreWarmedConnectionCount:Int
 
 init(
  idGenerator:Connection.ID.Generator,
  maximumConcurrentConnections:Int,
  retryConnectionEstablishment:Bool,
  maximumConnectionUses:Int?,
+ preWarmedHTTP1ConnectionCount:Int,
  lifecycleState:StateMachine.LifecycleState
 ){
 self.connections =HTTP1Connections(
  maximumConcurrentConnections: maximumConcurrentConnections,
  generator: idGenerator,
- maximumConnectionUses: maximumConnectionUses
+ maximumConnectionUses: maximumConnectionUses,
+ preWarmedHTTP1ConnectionCount: preWarmedHTTP1ConnectionCount
 )
+self.preWarmedConnectionCount = preWarmedHTTP1ConnectionCount
 self.retryConnectionEstablishment = retryConnectionEstablishment
 
 self.requests =RequestQueue()
@@ -145,9 +149,26 @@ extension HTTPConnectionPool{
 
 privatemutatingfunc executeRequestOnPreferredEventLoop(_ request:Request, eventLoop:EventLoop)->Action{
 iflet connection =self.connections.leaseConnection(onPreferred: eventLoop){
+ // Cool, a connection is available. If using this would put us below our needed extra set, we
+ // should create another.
+letstats=self.connections.generalPurposeStats
+letneedExtraConnection=
+ stats.nonLeased <(self.requests.count +self.preWarmedConnectionCount) && self.connections.canGrow
+letaction:StateMachine.ConnectionAction
+
+if needExtraConnection {
+ action =.createConnectionAndCancelTimeoutTimer(
+ createdID:self.connections.createNewConnection(on: eventLoop),
+ on: eventLoop,
+ cancelTimerID: connection.id
+)
+}else{
+ action =.cancelTimeoutTimer(connection.id)
+}
+
 return.init(
  request:.executeRequest(request, connection, cancelTimeout:false),
- connection:.cancelTimeoutTimer(connection.id)
+ connection:action
 )
 }
 
@@ -294,7 +315,20 @@ extension HTTPConnectionPool{
 }
 }
 
-mutatingfunc connectionIdleTimeout(_ connectionID:Connection.ID)->Action{
+mutatingfunc connectionIdleTimeout(_ connectionID:Connection.ID, on eventLoop:anyEventLoop)->Action{
+ // Don't close idle connections if we need pre-warmed connections. Instead, re-arm the idle timer.
+ // We still want the idle timers to make sure we eventually fall below the pre-warmed limit.
+ifself.preWarmedConnectionCount >0{
+letstats=self.connections.generalPurposeStats
+if stats.idle <=self.preWarmedConnectionCount {
+return.init(
+ request:.none,
+ connection:.scheduleTimeoutTimer(connectionID, on: eventLoop)
+)
+}
+}
+
+ // Ok, we do actually want the connection count to go down.
 guardlet connection =self.connections.closeConnectionIfIdle(connectionID)else{
  // because of a race this connection (connection close runs against trigger of timeout)
  // was already removed from the state machine.
@@ -410,11 +444,7 @@ extension HTTPConnectionPool{
 case.running:
  // Close the connection if it's expired.
 if context.shouldBeClosed {
-letconnection=self.connections.closeConnection(at: index)
-return.init(
- request:.none,
- connection:.closeConnection(connection, isShutdown:.no)
-)
+returnself.nextActionForToBeClosedIdleConnection(at: index, context: context)
 }else{
 switch context.use {
 case.generalPurpose:
@@ -446,28 +476,63 @@ extension HTTPConnectionPool{
  at index:Int,
  context:HTTP1Connections.IdleConnectionContext
 )->EstablishedAction{
+varrequestAction=HTTPConnectionPool.StateMachine.RequestAction.none
+varparkedConnectionDetails:(HTTPConnectionPool.Connection.ID,anyEventLoop)?=nil
+
  // 1. Check if there are waiting requests in the general purpose queue
 iflet request =self.requests.popFirst(for:nil){
-return.init(
- request:.executeRequest(request,self.connections.leaseConnection(at: index), cancelTimeout:true),
- connection:.none
+ requestAction =.executeRequest(
+ request,
+self.connections.leaseConnection(at: index),
+ cancelTimeout:true
 )
 }
 
  // 2. Check if there are waiting requests in the matching eventLoop queue
-iflet request =self.requests.popFirst(for: context.eventLoop){
-return.init(
- request:.executeRequest(request,self.connections.leaseConnection(at: index), cancelTimeout:true),
- connection:.none
+if case .none = requestAction,let request =self.requests.popFirst(for: context.eventLoop){
+ requestAction =.executeRequest(
+ request,
+self.connections.leaseConnection(at: index),
+ cancelTimeout:true
 )
 }
 
  // 3. Create a timeout timer to ensure the connection is closed if it is idle for too
- // long.
-let(connectionID, eventLoop)=self.connections.parkConnection(at: index)
+ // long, assuming we don't already have a use for it.
+if case .none = requestAction {
+ parkedConnectionDetails =self.connections.parkConnection(at: index)
+}
+
+ // 4. We may need to create another connection to make sure we have enough pre-warmed ones.
+ // We need to do that if we have fewer non-leased connections than we need pre-warmed ones _and_ the pool can grow.
+ // Note that in this case we don't need to account for the number of pending requests, as that is 0: step 1
+ // confirmed that.
+letconnectionAction:EstablishedConnectionAction
+
+ifself.connections.generalPurposeStats.nonLeased <self.preWarmedConnectionCount
+ && self.connections.canGrow
+{
+ // Re-use the event loop of the connection that just got created.
+iflet parkedConnectionDetails {
+letnewConnectionID=self.connections.createNewConnection(on: parkedConnectionDetails.1)
+ connectionAction =.scheduleTimeoutTimerAndCreateConnection(
+ timeoutID: parkedConnectionDetails.0,
+ newConnectionID: newConnectionID,
+ on: parkedConnectionDetails.1
+)
+}else{
+letnewConnectionID=self.connections.createNewConnection(on: context.eventLoop)
+ connectionAction =.createConnection(connectionID: newConnectionID, on: context.eventLoop)
+}
+}elseiflet parkedConnectionDetails {
+ connectionAction =.scheduleTimeoutTimer(parkedConnectionDetails.0, on: parkedConnectionDetails.1)
+}else{
+ connectionAction =.none
+}
+
 return.init(
- request:.none,
- connection:.scheduleTimeoutTimer(connectionID, on: eventLoop)
+ request:requestAction,
+ connection:connectionAction
 )
 }
 
@@ -495,6 +560,37 @@ extension HTTPConnectionPool{
 )
 }
 
+privatemutatingfunc nextActionForToBeClosedIdleConnection(
+ at index:Int,
+ context:HTTP1Connections.IdleConnectionContext
+)->EstablishedAction{
+ // Step 1: Tell the connection pool to drop what it knows about this object.
+letconnectionToClose=self.connections.closeConnection(at: index)
+
+ // Step 2: Check whether we need a connection to replace this one. We do if we have fewer non-leased connections
+ // than we requests + minimumPrewarming count _and_ the pool can grow. Note that in many cases the above closure
+ // will have made some space, which is just fine.
+letnonLeased=self.connections.generalPurposeStats.nonLeased
+letneededNonLeased=self.requests.generalPurposeCount +self.preWarmedConnectionCount
+
+letconnectionAction:EstablishedConnectionAction
+if nonLeased < neededNonLeased && self.connections.canGrow {
+ // We re-use the EL of the connection we just closed.
+letnewConnectionID=self.connections.createNewConnection(on: connectionToClose.eventLoop)
+ connectionAction =.closeConnectionAndCreateConnection(
+ closeConnection: connectionToClose,
+ newConnectionID: newConnectionID,
+ on: connectionToClose.eventLoop
+)
+}else{
+ connectionAction =.closeConnection(connectionToClose, isShutdown:.no)
+}
+return.init(
+ request:.none,
+ connection: connectionAction
+)
+}
+
  // MARK: Failed/Closed connection management
 
 privatemutatingfunc nextActionForFailedConnection(
@@ -530,7 +626,10 @@ extension HTTPConnectionPool{
  at index:Int,
  context:HTTP1Connections.FailedConnectionContext
 )->Action{
-if context.connectionsStartingForUseCase <self.requests.generalPurposeCount {
+letneedConnectionForRequest=
+ context.connectionsStartingForUseCase
+<(self.requests.generalPurposeCount +self.preWarmedConnectionCount)
+if needConnectionForRequest {
  // if we have more requests queued up, than we have starting connections, we should
  // create a new connection
 let(newConnectionID, newEventLoop)=self.connections.replaceConnection(at: index)