Skip to content

Commit c1a60d8

Browse files
dnadobadnadoba
authored
[HTTP2] Prepare migration actions (#456)
1 parent 1081b0b commit c1a60d8

12 files changed

+547
-93
lines changed

‎.swiftformat‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,6 @@
99
--patternlet inline
1010
--stripunusedargs unnamed-only
1111
--ranges nospace
12-
--disable typeSugar # https://github.com/nicklockwood/SwiftFormat/issues/636
12+
--disable typeSugar, andOperator # typeSugar: https://github.com/nicklockwood/SwiftFormat/issues/636
1313

1414
# rules

‎Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,6 @@ final class HTTPConnectionPool {
6969
self.idleConnectionTimeout = clientConfiguration.connectionPool.idleTimeout
7070

7171
self._state = StateMachine(
72-
eventLoopGroup: eventLoopGroup,
7372
idGenerator: idGenerator,
7473
maximumConcurrentHTTP1Connections: clientConfiguration.connectionPool.concurrentHTTP1ConnectionsPerHostSoftLimit
7574
)
@@ -97,6 +96,10 @@ final class HTTPConnectionPool {
9796
case createConnection(Connection.ID, on: EventLoop)
9897
case closeConnection(Connection, isShutdown: StateMachine.ConnectionAction.IsShutdown)
9998
case cleanupConnections(CleanupContext, isShutdown: StateMachine.ConnectionAction.IsShutdown)
99+
case migration(
100+
createConnections: [(Connection.ID, EventLoop)],
101+
closeConnections: [Connection]
102+
)
100103
case none
101104
}
102105

@@ -184,6 +187,18 @@ final class HTTPConnectionPool {
184187
self.locked.connection = .cancelBackoffTimers(cleanupContext.connectBackoff)
185188
cleanupContext.connectBackoff = []
186189
self.unlocked.connection = .cleanupConnections(cleanupContext, isShutdown: isShutdown)
190+
case .migration(
191+
let createConnections,
192+
let closeConnections,
193+
let scheduleTimeout
194+
):
195+
if let (connectionID, eventLoop) = scheduleTimeout {
196+
self.locked.connection = .scheduleTimeoutTimer(connectionID, on: eventLoop)
197+
}
198+
self.unlocked.connection = .migration(
199+
createConnections: createConnections,
200+
closeConnections: closeConnections
201+
)
187202
case .none:
188203
break
189204
}
@@ -279,6 +294,15 @@ final class HTTPConnectionPool {
279294
self.delegate.connectionPoolDidShutdown(self, unclean: unclean)
280295
}
281296

297+
case .migration(let createConnections, let closeConnections):
298+
for connection in closeConnections {
299+
connection.close(promise: nil)
300+
}
301+
302+
for (connectionID, eventLoop) in createConnections {
303+
self.createConnection(connectionID, on: eventLoop)
304+
}
305+
282306
case .none:
283307
break
284308
}

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

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -530,6 +530,28 @@ extension HTTPConnectionPool {
530530
return migrationContext
531531
}
532532

533+
/// we only handle starting and backing off connection here.
534+
/// All running connections must be handled by the enclosing state machine
535+
/// - Parameters:
536+
/// - starting: starting HTTP connections from previous state machine
537+
/// - backingOff: backing off HTTP connections from previous state machine
538+
mutating func migrateFromHTTP2(
539+
starting: [(Connection.ID, EventLoop)],
540+
backingOff: [(Connection.ID, EventLoop)]
541+
) {
542+
for (connectionID, eventLoop) in starting {
543+
let newConnection = HTTP1ConnectionState(connectionID: connectionID, eventLoop: eventLoop)
544+
self.connections.append(newConnection)
545+
}
546+
547+
for (connectionID, eventLoop) in backingOff {
548+
var backingOffConnection = HTTP1ConnectionState(connectionID: connectionID, eventLoop: eventLoop)
549+
// TODO: Maybe we want to add a static init for backing off connections to HTTP1ConnectionState
550+
backingOffConnection.failedToConnect()
551+
self.connections.append(backingOffConnection)
552+
}
553+
}
554+
533555
// MARK: Shutdown
534556

535557
mutating func shutdown() -> CleanupContext {

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

Lines changed: 175 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,12 @@ extension HTTPConnectionPool {
2323
}
2424

2525
typealias Action = HTTPConnectionPool.StateMachine.Action
26+
typealias ConnectionMigrationAction = HTTPConnectionPool.StateMachine.ConnectionMigrationAction
27+
typealias EstablishedAction = HTTPConnectionPool.StateMachine.EstablishedAction
28+
typealias EstablishedConnectionAction = HTTPConnectionPool.StateMachine.EstablishedConnectionAction
2629

2730
private(set) var connections: HTTP1Connections
31+
private(set) var http2Connections: HTTP2Connections?
2832
private var failedConsecutiveConnectionAttempts: Int = 0
2933
/// the error from the last connection creation
3034
private var lastConnectFailure: Error?
@@ -41,6 +45,73 @@ extension HTTPConnectionPool {
4145
self.requests = RequestQueue()
4246
}
4347

48+
mutating func migrateFromHTTP2(
49+
http2State: HTTP2StateMachine,
50+
newHTTP1Connection: Connection
51+
) -> Action {
52+
self.migrateFromHTTP2(
53+
http1Connections: http2State.http1Connections,
54+
http2Connections: http2State.connections,
55+
requests: http2State.requests,
56+
newHTTP1Connection: newHTTP1Connection
57+
)
58+
}
59+
60+
mutating func migrateFromHTTP2(
61+
http1Connections: HTTP1Connections? = nil,
62+
http2Connections: HTTP2Connections,
63+
requests: RequestQueue,
64+
newHTTP1Connection: Connection
65+
) -> Action {
66+
let migrationAction = self.migrateConnectionsAndRequestsFromHTTP2(
67+
http1Connections: http1Connections,
68+
http2Connections: http2Connections,
69+
requests: requests
70+
)
71+
72+
let newConnectionAction = self._newHTTP1ConnectionEstablished(
73+
newHTTP1Connection
74+
)
75+
76+
return .init(
77+
request: newConnectionAction.request,
78+
connection: .combined(migrationAction, newConnectionAction.connection)
79+
)
80+
}
81+
82+
private mutating func migrateConnectionsAndRequestsFromHTTP2(
83+
http1Connections: HTTP1Connections?,
84+
http2Connections: HTTP2Connections,
85+
requests: RequestQueue
86+
) -> ConnectionMigrationAction {
87+
precondition(self.connections.isEmpty)
88+
precondition(self.http2Connections == nil)
89+
precondition(self.requests.isEmpty)
90+
91+
if let http1Connections = http1Connections {
92+
self.connections = http1Connections
93+
}
94+
95+
var http2Connections = http2Connections
96+
let migration = http2Connections.migrateToHTTP1()
97+
self.connections.migrateFromHTTP2(
98+
starting: migration.starting,
99+
backingOff: migration.backingOff
100+
)
101+
102+
if !http2Connections.isEmpty {
103+
self.http2Connections = http2Connections
104+
}
105+
106+
// TODO: Close all idle connections from context.close
107+
// TODO: Start new http1 connections for pending requests
108+
// TODO: Potentially cancel unneeded bootstraps (Needs cancellable ClientBootstrap)
109+
110+
self.requests = requests
111+
112+
return .init(closeConnections: [], createConnections: [])
113+
}
114+
44115
// MARK: - Events
45116

46117
mutating func executeRequest(_ request: Request) -> Action {
@@ -137,6 +208,10 @@ extension HTTPConnectionPool {
137208
}
138209

139210
mutating func newHTTP1ConnectionEstablished(_ connection: Connection) -> Action {
211+
.init(self._newHTTP1ConnectionEstablished(connection))
212+
}
213+
214+
private mutating func _newHTTP1ConnectionEstablished(_ connection: Connection) -> EstablishedAction {
140215
self.failedConsecutiveConnectionAttempts = 0
141216
self.lastConnectFailure = nil
142217
let (index, context) = self.connections.newHTTP1ConnectionEstablished(connection)
@@ -210,7 +285,7 @@ extension HTTPConnectionPool {
210285

211286
mutating func http1ConnectionReleased(_ connectionID: Connection.ID) -> Action {
212287
let (index, context) = self.connections.releaseConnection(connectionID)
213-
return self.nextActionForIdleConnection(at: index, context: context)
288+
return .init(self.nextActionForIdleConnection(at: index, context: context))
214289
}
215290

216291
/// A connection has been unexpectedly closed
@@ -278,7 +353,7 @@ extension HTTPConnectionPool {
278353
// If there aren't any more connections, everything is shutdown
279354
let isShutdown: StateMachine.ConnectionAction.IsShutdown
280355
let unclean = !(cleanupContext.cancel.isEmpty && waitingRequests.isEmpty)
281-
if self.connections.isEmpty {
356+
if self.connections.isEmpty && self.http2Connections == nil {
282357
self.state = .shutDown
283358
isShutdown = .yes(unclean: unclean)
284359
} else {
@@ -299,7 +374,7 @@ extension HTTPConnectionPool {
299374
private mutating func nextActionForIdleConnection(
300375
at index: Int,
301376
context: HTTP1Connections.IdleConnectionContext
302-
) -> Action {
377+
) -> EstablishedAction {
303378
switch self.state {
304379
case .running:
305380
switch context.use {
@@ -311,7 +386,7 @@ extension HTTPConnectionPool {
311386
case .shuttingDown(let unclean):
312387
assert(self.requests.isEmpty)
313388
let connection = self.connections.closeConnection(at: index)
314-
if self.connections.isEmpty {
389+
if self.connections.isEmpty && self.http2Connections == nil {
315390
return .init(
316391
request: .none,
317392
connection: .closeConnection(connection, isShutdown: .yes(unclean: unclean))
@@ -330,7 +405,7 @@ extension HTTPConnectionPool {
330405
private mutating func nextActionForIdleGeneralPurposeConnection(
331406
at index: Int,
332407
context: HTTP1Connections.IdleConnectionContext
333-
) -> Action {
408+
) -> EstablishedAction {
334409
// 1. Check if there are waiting requests in the general purpose queue
335410
if let request = self.requests.popFirst(for: nil) {
336411
return .init(
@@ -359,7 +434,7 @@ extension HTTPConnectionPool {
359434
private mutating func nextActionForIdleEventLoopConnection(
360435
at index: Int,
361436
context: HTTP1Connections.IdleConnectionContext
362-
) -> Action {
437+
) -> EstablishedAction {
363438
// Check if there are waiting requests in the matching eventLoop queue
364439
if let request = self.requests.popFirst(for: context.eventLoop) {
365440
return .init(
@@ -398,7 +473,7 @@ extension HTTPConnectionPool {
398473
case .shuttingDown(let unclean):
399474
assert(self.requests.isEmpty)
400475
self.connections.removeConnection(at: index)
401-
if self.connections.isEmpty {
476+
if self.connections.isEmpty && self.http2Connections == nil {
402477
return .init(
403478
request: .none,
404479
connection: .cleanupConnections(.init(), isShutdown: .yes(unclean: unclean))
@@ -444,6 +519,99 @@ extension HTTPConnectionPool {
444519
self.connections.removeConnection(at: index)
445520
return .none
446521
}
522+
523+
// MARK: HTTP2
524+
525+
mutating func newHTTP2MaxConcurrentStreamsReceived(_ connectionID: Connection.ID, newMaxStreams: Int) -> Action {
526+
// It is save to bang the http2Connections here. If we get this callback but we don't have
527+
// http2 connections something has gone terribly wrong.
528+
_ = self.http2Connections!.newHTTP2MaxConcurrentStreamsReceived(connectionID, newMaxStreams: newMaxStreams)
529+
return .none
530+
}
531+
532+
mutating func http2ConnectionGoAwayReceived(_ connectionID: Connection.ID) -> Action {
533+
// It is save to bang the http2Connections here. If we get this callback but we don't have
534+
// http2 connections something has gone terribly wrong.
535+
_ = self.http2Connections!.goAwayReceived(connectionID)
536+
return .none
537+
}
538+
539+
mutating func http2ConnectionClosed(_ connectionID: Connection.ID) -> Action {
540+
switch self.state {
541+
case .running:
542+
_ = self.http2Connections?.failConnection(connectionID)
543+
if self.http2Connections?.isEmpty == true {
544+
self.http2Connections = nil
545+
}
546+
return .none
547+
548+
case .shuttingDown(let unclean):
549+
assert(self.requests.isEmpty)
550+
_ = self.http2Connections?.failConnection(connectionID)
551+
if self.http2Connections?.isEmpty == true {
552+
self.http2Connections = nil
553+
}
554+
if self.connections.isEmpty && self.http2Connections == nil {
555+
return .init(
556+
request: .none,
557+
connection: .cleanupConnections(.init(), isShutdown: .yes(unclean: unclean))
558+
)
559+
}
560+
return .init(
561+
request: .none,
562+
connection: .none
563+
)
564+
565+
case .shutDown:
566+
preconditionFailure("It the pool is already shutdown, all connections must have been torn down.")
567+
}
568+
}
569+
570+
mutating func http2ConnectionStreamClosed(_ connectionID: Connection.ID) -> Action {
571+
// It is save to bang the http2Connections here. If we get this callback but we don't have
572+
// http2 connections something has gone terribly wrong.
573+
switch self.state {
574+
case .running:
575+
let (index, context) = self.http2Connections!.releaseStream(connectionID)
576+
guard context.isIdle else {
577+
return .none
578+
}
579+
580+
let connection = self.http2Connections!.closeConnection(at: index)
581+
if self.http2Connections!.isEmpty {
582+
self.http2Connections = nil
583+
}
584+
return .init(
585+
request: .none,
586+
connection: .closeConnection(connection, isShutdown: .no)
587+
)
588+
589+
case .shuttingDown(let unclean):
590+
assert(self.requests.isEmpty)
591+
let (index, context) = self.http2Connections!.releaseStream(connectionID)
592+
guard context.isIdle else {
593+
return .none
594+
}
595+
596+
let connection = self.http2Connections!.closeConnection(at: index)
597+
if self.http2Connections!.isEmpty {
598+
self.http2Connections = nil
599+
}
600+
if self.connections.isEmpty && self.http2Connections == nil {
601+
return .init(
602+
request: .none,
603+
connection: .closeConnection(connection, isShutdown: .yes(unclean: unclean))
604+
)
605+
}
606+
return .init(
607+
request: .none,
608+
connection: .closeConnection(connection, isShutdown: .no)
609+
)
610+
611+
case .shutDown:
612+
preconditionFailure("It the pool is already shutdown, all connections must have been torn down.")
613+
}
614+
}
447615
}
448616
}
449617

0 commit comments

Comments
 (0)