[ConnectionPool] Split up locked and unlocked actions

# Conflicts: # Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift

Author: fabianfett

Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift

@@ -131,15 +131,17 @@ final class HTTPConnectionPool{
 
 privateletstateLock=Lock()
 privatevar_state:StateMachine
+ /// The connection idle timeout timers. Protected by the stateLock
+privatevar_idleTimer=[Connection.ID:Scheduled<Void>]()
+ /// The connection backoff timeout timers. Protected by the stateLock
+privatevar_backoffTimer=[Connection.ID:Scheduled<Void>]()
 
 privatestaticletfallbackConnectTimeout:TimeAmount=.seconds(30)
 
 letkey:ConnectionPool.Key
 
 privatelettimerLock=Lock()
 privatevar_requestTimer=[Request.ID:Scheduled<Void>]()
-privatevar_idleTimer=[Connection.ID:Scheduled<Void>]()
-privatevar_backoffTimer=[Connection.ID:Scheduled<Void>]()
 
 privatevarlogger:Logger
 
@@ -182,33 +184,90 @@ final class HTTPConnectionPool{
 }
 
 func executeRequest(_ request:HTTPSchedulableRequest){
-letaction=self.stateLock.withLock{()->StateMachine.Actionin
-self._state.executeRequest(.init(request))
-}
-self.run(action: action)
+self.modifyStateAndRunActions{ $0.executeRequest(.init(request))}
 }
 
 func shutdown(){
 self.logger.debug("Shutting down connection pool")
-letaction=self.stateLock.withLock{()->StateMachine.Actionin
-self._state.shutdown()
+self.modifyStateAndRunActions{ $0.shutdown()}
+}
+
+ // MARK: - Private Methods -
+
+ // MARK: Actions
+
+ ///
+privatestructActions{
+enumConnectionAction{
+enumUnlocked{
+case createConnection(Connection.ID, on:EventLoop)
+case closeConnection(Connection, isShutdown:StateMachine.ConnectionAction.IsShutdown)
+case cleanupConnections(CleanupContext, isShutdown:StateMachine.ConnectionAction.IsShutdown)
+case none
+}
+
+enumLocked{
+case scheduleBackoffTimer(Connection.ID, backoff:TimeAmount, on:EventLoop)
+case cancelBackoffTimers([Connection.ID])
+case scheduleTimeoutTimer(Connection.ID, on:EventLoop)
+case cancelTimeoutTimer(Connection.ID)
+case none
+}
+}
+
+structLocked{
+varconnection:ConnectionAction.Locked
+}
+
+structUnlocked{
+varconnection:ConnectionAction.Unlocked
+varrequest:StateMachine.RequestAction
+}
+
+varlocked:Locked
+varunlocked:Unlocked
+
+init(from stateMachineAction:StateMachine.Action){
+self.locked =Locked(connection:.none)
+self.unlocked =Unlocked(connection:.none, request: stateMachineAction.request)
+
+switch stateMachineAction.connection {
+case.createConnection(let connectionID, on:let eventLoop):
+self.unlocked.connection =.createConnection(connectionID, on: eventLoop)
+case.scheduleBackoffTimer(let connectionID, backoff:let backoff, on:let eventLoop):
+self.locked.connection =.scheduleBackoffTimer(connectionID, backoff: backoff, on: eventLoop)
+case.scheduleTimeoutTimer(let connectionID, on:let eventLoop):
+self.locked.connection =.scheduleTimeoutTimer(connectionID, on: eventLoop)
+case.cancelTimeoutTimer(let connectionID):
+self.locked.connection =.cancelTimeoutTimer(connectionID)
+case.closeConnection(let connection, isShutdown:let isShutdown):
+self.unlocked.connection =.closeConnection(connection, isShutdown: isShutdown)
+case.cleanupConnections(var cleanupContext, isShutdown:let isShutdown):
+ //
+self.locked.connection =.cancelBackoffTimers(cleanupContext.connectBackoff)
+ cleanupContext.connectBackoff =[]
+self.unlocked.connection =.cleanupConnections(cleanupContext, isShutdown: isShutdown)
+case.none:
+break
+}
 }
-self.run(action: action)
 }
 
  // MARK: Run actions
 
-privatefunc run(action:StateMachine.Action){
-self.runConnectionAction(action.connection)
-self.runRequestAction(action.request)
+privatefunc modifyStateAndRunActions(_ closure:(inoutStateMachine)->StateMachine.Action){
+letunlockedActions=self.stateLock.withLock{()->Actions.Unlockedin
+letstateMachineAction=closure(&self._state)
+letpoolAction=Actions(from: stateMachineAction)
+self.runLockedActions(poolAction.locked)
+return poolAction.unlocked
+}
+self.runUnlockedActions(unlockedActions)
 }
 
-privatefunc runConnectionAction(_ action:StateMachine.ConnectionAction){
-switch action {
-case.createConnection(let connectionID,let eventLoop):
-self.createConnection(connectionID, on: eventLoop)
-
-case.scheduleBackoffTimer(let connectionID,let backoff, on:let eventLoop):
+privatefunc runLockedActions(_ actions:Actions.Locked){
+switch actions.connection {
+case.scheduleBackoffTimer(let connectionID, backoff:let backoff, on:let eventLoop):
 self.scheduleConnectionStartBackoffTimer(connectionID, backoff, on: eventLoop)
 
 case.scheduleTimeoutTimer(let connectionID, on:let eventLoop):
@@ -217,6 +276,26 @@ final class HTTPConnectionPool{
 case.cancelTimeoutTimer(let connectionID):
 self.cancelIdleTimerForConnection(connectionID)
 
+case.cancelBackoffTimers(let connectionIDs):
+forconnectionIDin connectionIDs {
+self.cancelConnectionStartBackoffTimer(connectionID)
+}
+
+case.none:
+break
+}
+}
+
+privatefunc runUnlockedActions(_ actions:Actions.Unlocked){
+self.runUnlockedConnectionAction(actions.connection)
+self.runUnlockedRequestAction(actions.request)
+}
+
+privatefunc runUnlockedConnectionAction(_ action:Actions.ConnectionAction.Unlocked){
+switch action {
+case.createConnection(let connectionID,let eventLoop):
+self.createConnection(connectionID, on: eventLoop)
+
 case.closeConnection(let connection, isShutdown:let isShutdown):
 self.logger.trace("close connection", metadata:[
 "ahc-connection-id":"\(connection.id)",
@@ -251,7 +330,7 @@ final class HTTPConnectionPool{
 }
 }
 
-privatefuncrunRequestAction(_ action:StateMachine.RequestAction){
+privatefuncrunUnlockedRequestAction(_ action:StateMachine.RequestAction){
  // The order of execution fail/execute request vs cancelling the request timeout timer does
  // not matter in the actions here. The actions don't cause any side effects that will be
  // reported back to the state machine and are not dependent on each other.
@@ -323,11 +402,9 @@ final class HTTPConnectionPool{
 guard timeoutFired else{return}
 
  // 3. Tell the state machine about the timeout
-letaction=self.stateLock.withLock{
-self._state.timeoutRequest(requestID)
+self.modifyStateAndRunActions{
+$0.timeoutRequest(requestID)
 }
-
-self.run(action: action)
 }
 
 self.timerLock.withLockVoid{
@@ -362,34 +439,27 @@ final class HTTPConnectionPool{
 letscheduled= eventLoop.scheduleTask(in:self.idleConnectionTimeout){
  // there might be a race between a cancelTimer call and the triggering
  // of this scheduled task. both want to acquire the lock
-lettimerExisted=self.timerLock.withLock{
-self._idleTimer.removeValue(forKey: connectionID)!=nil
+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.none
 }
-
-guard timerExisted else{return}
-
-letaction=self.stateLock.withLock{
-self._state.connectionIdleTimeout(connectionID)
-}
-self.run(action: action)
 }
 
-self.timerLock.withLock{
-assert(self._idleTimer[connectionID]==nil)
-self._idleTimer[connectionID]= scheduled
-}
+assert(self._idleTimer[connectionID]==nil)
+self._idleTimer[connectionID]= scheduled
 }
 
 privatefunc cancelIdleTimerForConnection(_ connectionID:Connection.ID){
 self.logger.trace("Cancel idle connection timeout timer", metadata:[
 "ahc-connection-id":"\(connectionID)",
 ])
-
-letcancelTimer=self.timerLock.withLock{
-self._idleTimer.removeValue(forKey: connectionID)
+guardlet cancelTimer =self._idleTimer.removeValue(forKey: connectionID)else{
+preconditionFailure("Expected to have an idle timer for connection \(connectionID) at this point.")
 }
-
- cancelTimer?.cancel()
+ cancelTimer.cancel()
 }
 
 privatefunc scheduleConnectionStartBackoffTimer(
@@ -403,30 +473,24 @@ final class HTTPConnectionPool{
 
 letscheduled= eventLoop.scheduleTask(in: timeAmount){
  // there might be a race between a backoffTimer and the pool shutting down.
-lettimerExisted=self.timerLock.withLock{
-self._backoffTimer.removeValue(forKey: connectionID)!=nil
-}
-
-guard timerExisted else{return}
-
-letaction=self.stateLock.withLock{
-self._state.connectionCreationBackoffDone(connectionID)
+self.modifyStateAndRunActions{ stateMachine in
+ifself._backoffTimer.removeValue(forKey: connectionID)!=nil{
+ // The timer still exists. State Machines assumes it is alive
+return stateMachine.connectionCreationBackoffDone(connectionID)
+}
+return.none
 }
-self.run(action: action)
 }
 
-self.timerLock.withLock{
-assert(self._backoffTimer[connectionID]==nil)
-self._backoffTimer[connectionID]= scheduled
-}
+assert(self._backoffTimer[connectionID]==nil)
+self._backoffTimer[connectionID]= scheduled
 }
 
 privatefunc cancelConnectionStartBackoffTimer(_ connectionID:Connection.ID){
-letbackoffTimer=self.timerLock.withLock{
-self._backoffTimer[connectionID]
+guardlet backoffTimer =self._backoffTimer.removeValue(forKey: connectionID)else{
+preconditionFailure("Expected to have a backoff timer for connection \(connectionID) at this point.")
 }
-
- backoffTimer?.cancel()
+ backoffTimer.cancel()
 }
 }
 
@@ -438,10 +502,9 @@ extension HTTPConnectionPool: HTTPConnectionRequester{
 "ahc-connection-id":"\(connection.id)",
 "ahc-http-version":"http/1.1",
 ])
-letaction=self.stateLock.withLock{
-self._state.newHTTP1ConnectionCreated(.http1_1(connection))
+self.modifyStateAndRunActions{
+$0.newHTTP1ConnectionCreated(.http1_1(connection))
 }
-self.run(action: action)
 }
 
 func http2ConnectionCreated(_ connection:HTTP2Connection, maximumStreams:Int){
@@ -464,10 +527,9 @@ extension HTTPConnectionPool: HTTPConnectionRequester{
 "ahc-error":"\(error)",
 "ahc-connection-id":"\(connectionID)",
 ])
-letaction=self.stateLock.withLock{
-self._state.failedToCreateNewConnection(error, connectionID: connectionID)
+self.modifyStateAndRunActions{
+$0.failedToCreateNewConnection(error, connectionID: connectionID)
 }
-self.run(action: action)
 }
 }
 
@@ -477,21 +539,19 @@ extension HTTPConnectionPool: HTTP1ConnectionDelegate{
 "ahc-connection-id":"\(connection.id)",
 "ahc-http-version":"http/1.1",
 ])
-letaction=self.stateLock.withLock{
-self._state.connectionClosed(connection.id)
+self.modifyStateAndRunActions{
+$0.connectionClosed(connection.id)
 }
-self.run(action: action)
 }
 
 func http1ConnectionReleased(_ connection:HTTP1Connection){
 self.logger.trace("releasing connection", metadata:[
 "ahc-connection-id":"\(connection.id)",
 "ahc-http-version":"http/1.1",
 ])
-letaction=self.stateLock.withLock{
-self._state.http1ConnectionReleased(connection.id)
+self.modifyStateAndRunActions{
+$0.http1ConnectionReleased(connection.id)
 }
-self.run(action: action)
 }
 }
 
@@ -524,10 +584,9 @@ extension HTTPConnectionPool: HTTP2ConnectionDelegate{
 extensionHTTPConnectionPool:HTTPRequestScheduler{
 func cancelRequest(_ request:HTTPSchedulableRequest){
 letrequestID=Request(request).id
-letaction=self.stateLock.withLock{
-self._state.cancelRequest(requestID)
+self.modifyStateAndRunActions{
+$0.cancelRequest(requestID)
 }
-self.run(action: action)
 }
 }