13
13
//===----------------------------------------------------------------------===//
14
14
15
15
import Foundation
16
+ import Logging
16
17
import NIO
17
18
import NIOConcurrencyHelpers
18
19
import NIOHTTP1
@@ -64,21 +65,29 @@ final class ConnectionPool {
64
65
/// When the pool is asked for a new connection, it creates a `Key` from the url associated to the `request`. This key
65
66
/// is used to determine if there already exists an associated `HTTP1ConnectionProvider` in `providers`.
66
67
/// If there is, the connection provider then takes care of leasing a new connection. If a connection provider doesn't exist, it is created.
67
- func getConnection( for request: HTTPClient . Request , preference: HTTPClient . EventLoopPreference , on eventLoop: EventLoop , deadline: NIODeadline ? , setupComplete: EventLoopFuture < Void > ) -> EventLoopFuture < Connection > {
68
+ func getConnection( _ request: HTTPClient . Request ,
69
+ preference: HTTPClient . EventLoopPreference ,
70
+ taskEventLoop: EventLoop ,
71
+ deadline: NIODeadline ? ,
72
+ setupComplete: EventLoopFuture < Void > ,
73
+ logger: Logger ) -> EventLoopFuture < Connection > {
68
74
let key = Key ( request)
69
75
70
76
let provider : HTTP1ConnectionProvider = self . lock. withLock {
71
77
if let existing = self . providers [ key] , existing. enqueue ( ) {
72
78
return existing
73
79
} else {
74
80
// Connection provider will be created with `pending = 1`
75
- let provider = HTTP1ConnectionProvider ( key: key, eventLoop: eventLoop, configuration: self . configuration, pool: self )
81
+ let provider = HTTP1ConnectionProvider ( key: key,
82
+ eventLoop: taskEventLoop,
83
+ configuration: self . configuration,
84
+ pool: self )
76
85
self . providers [ key] = provider
77
86
return provider
78
87
}
79
88
}
80
89
81
- return provider. getConnection ( preference: preference, setupComplete: setupComplete)
90
+ return provider. getConnection ( preference: preference, setupComplete: setupComplete, logger : logger )
82
91
}
83
92
84
93
func delete( _ provider: HTTP1ConnectionProvider ) {
@@ -167,21 +176,21 @@ class Connection {
167
176
/// Release this `Connection` to its associated `HTTP1ConnectionProvider`.
168
177
///
169
178
/// - Warning: This only releases the connection and doesn't take care of cleaning handlers in the `Channel` pipeline.
170
- func release( closing: Bool ) {
179
+ func release( closing: Bool , logger : Logger ) {
171
180
assert ( self . channel. eventLoop. inEventLoop)
172
- self . provider. release ( connection: self , closing: closing)
181
+ self . provider. release ( connection: self , closing: closing, logger : logger )
173
182
}
174
183
175
184
/// Called when channel exceeds idle time in pool.
176
- func timeout( ) {
185
+ func timeout( logger : Logger ) {
177
186
assert ( self . channel. eventLoop. inEventLoop)
178
- self . provider. timeout ( connection: self )
187
+ self . provider. timeout ( connection: self , logger : logger )
179
188
}
180
189
181
190
/// Called when channel goes inactive while in the pool.
182
- func remoteClosed( ) {
191
+ func remoteClosed( logger : Logger ) {
183
192
assert ( self . channel. eventLoop. inEventLoop)
184
- self . provider. remoteClosed ( connection: self )
193
+ self . provider. remoteClosed ( connection: self , logger : logger )
185
194
}
186
195
187
196
func cancel( ) -> EventLoopFuture < Void > {
@@ -194,9 +203,10 @@ class Connection {
194
203
}
195
204
196
205
/// Sets idle timeout handler and channel inactivity listener.
197
- func setIdleTimeout( timeout: TimeAmount ? ) {
206
+ func setIdleTimeout( timeout: TimeAmount ? , logger : Logger ) {
198
207
_ = self . channel. pipeline. addHandler ( IdleStateHandler ( writeTimeout: timeout) , position: . first) . flatMap { _ in
199
- self . channel. pipeline. addHandler ( IdlePoolConnectionHandler ( connection: self ) )
208
+ self . channel. pipeline. addHandler ( IdlePoolConnectionHandler ( connection: self ,
209
+ logger: logger) )
200
210
}
201
211
}
202
212
@@ -280,50 +290,65 @@ class HTTP1ConnectionProvider {
280
290
self . state. assertInvariants ( )
281
291
}
282
292
283
- private func execute( _ action: Action ) {
293
+ private func execute( _ action: Action , logger : Logger ) {
284
294
switch action {
285
295
case . lease( let connection, let waiter) :
286
296
// if connection is became inactive, we create a new one.
287
297
connection. cancelIdleTimeout ( ) . whenComplete { _ in
288
298
if connection. isActiveEstimation {
299
+ logger. trace ( " leasing existing connection " )
289
300
waiter. promise. succeed ( connection)
290
301
} else {
302
+ logger. trace ( " opening fresh connection (found matching but inactive connection) " ,
303
+ metadata: [ " ahc-dead-connection " : " \( connection) " ] )
291
304
self . makeChannel ( preference: waiter. preference) . whenComplete { result in
292
- self . connect ( result, waiter: waiter, replacing: connection)
305
+ self . connect ( result, waiter: waiter, replacing: connection, logger : logger )
293
306
}
294
307
}
295
308
}
296
309
case . create( let waiter) :
310
+ logger. trace ( " opening fresh connection (no connections to reuse available) " )
297
311
self . makeChannel ( preference: waiter. preference) . whenComplete { result in
298
- self . connect ( result, waiter: waiter)
312
+ self . connect ( result, waiter: waiter, logger : logger )
299
313
}
300
314
case . replace( let connection, let waiter) :
315
+ logger. trace ( " replacing \( connection) , \( waiter) " )
301
316
connection. cancelIdleTimeout ( ) . flatMap {
302
317
connection. close ( )
303
318
} . whenComplete { _ in
319
+ logger. trace ( " opening fresh connection (replacing \( connection) , \( waiter) " )
304
320
self . makeChannel ( preference: waiter. preference) . whenComplete { result in
305
- self . connect ( result, waiter: waiter, replacing: connection)
321
+ self . connect ( result, waiter: waiter, replacing: connection, logger : logger )
306
322
}
307
323
}
308
324
case . park( let connection) :
309
- connection. setIdleTimeout ( timeout: self . configuration. maximumAllowedIdleTimeInConnectionPool)
325
+ logger. trace ( " parking connection " )
326
+ connection. setIdleTimeout ( timeout: self . configuration. maximumAllowedIdleTimeInConnectionPool,
327
+ logger: logger. detachingRequestInformation ( nil ) )
310
328
case . closeProvider:
329
+ logger. debug ( " closing provider \( self ) " )
311
330
self . closeAndDelete ( )
312
331
case . none:
313
332
break
314
333
case . parkAnd( let connection, let action) :
315
- connection. setIdleTimeout ( timeout: self . configuration. maximumAllowedIdleTimeInConnectionPool)
316
- self . execute ( action)
334
+ logger. trace ( " parking connection & doing further action " ,
335
+ metadata: [ " ahc-action " : " \( action) " ] )
336
+ connection. setIdleTimeout ( timeout: self . configuration. maximumAllowedIdleTimeInConnectionPool,
337
+ logger: logger. detachingRequestInformation ( nil ) )
338
+ self . execute ( action, logger: logger)
317
339
case . closeAnd( let connection, let action) :
340
+ logger. debug ( " closing \( connection) , doing \( action) " )
318
341
connection. channel. close ( promise: nil )
319
- self . execute ( action)
342
+ self . execute ( action, logger : logger )
320
343
case . cancel( let connection, let close) :
344
+ logger. debug ( " cancelling \( connection) , close= \( close) " )
321
345
connection. cancel ( ) . whenComplete { _ in
322
346
if close {
323
347
self . closeAndDelete ( )
324
348
}
325
349
}
326
350
case . fail( let waiter, let error) :
351
+ logger. debug ( " failing \( waiter) : \( error) " )
327
352
waiter. promise. fail ( error)
328
353
}
329
354
}
@@ -335,19 +360,24 @@ class HTTP1ConnectionProvider {
335
360
}
336
361
}
337
362
338
- func getConnection( preference: HTTPClient . EventLoopPreference , setupComplete: EventLoopFuture < Void > ) -> EventLoopFuture < Connection > {
363
+ func getConnection( preference: HTTPClient . EventLoopPreference ,
364
+ setupComplete: EventLoopFuture < Void > ,
365
+ logger: Logger ) -> EventLoopFuture < Connection > {
339
366
let waiter = Waiter ( promise: self . eventLoop. makePromise ( ) , setupComplete: setupComplete, preference: preference)
340
367
341
368
let action : Action = self . lock. withLock {
342
369
self . state. acquire ( waiter: waiter)
343
370
}
344
371
345
- self . execute ( action)
372
+ self . execute ( action, logger : logger )
346
373
347
374
return waiter. promise. futureResult
348
375
}
349
376
350
- func connect( _ result: Result < Channel , Error > , waiter: Waiter , replacing closedConnection: Connection ? = nil ) {
377
+ func connect( _ result: Result < Channel , Error > ,
378
+ waiter: Waiter ,
379
+ replacing closedConnection: Connection ? = nil ,
380
+ logger: Logger ) {
351
381
let action : Action
352
382
switch result {
353
383
case . success( let channel) :
@@ -366,11 +396,11 @@ class HTTP1ConnectionProvider {
366
396
waiter. promise. fail ( error)
367
397
}
368
398
waiter. setupComplete. whenComplete { _ in
369
- self . execute ( action)
399
+ self . execute ( action, logger : logger )
370
400
}
371
401
}
372
402
373
- func release( connection: Connection , closing: Bool ) {
403
+ func release( connection: Connection , closing: Bool , logger : Logger ) {
374
404
let action : Action = self . lock. withLock {
375
405
self . state. release ( connection: connection, closing: closing)
376
406
}
@@ -381,31 +411,31 @@ class HTTP1ConnectionProvider {
381
411
case . park, . closeProvider:
382
412
// Since both `.park` and `.deleteProvider` are terminal in terms of execution,
383
413
// we can execute them immediately
384
- self . execute ( action)
414
+ self . execute ( action, logger : logger )
385
415
case . cancel, . closeAnd, . create, . fail, . lease, . parkAnd, . replace:
386
416
// This is needed to start a new stack, otherwise, since this is called on a previous
387
417
// future completion handler chain, it will be growing indefinitely until the connection is closed.
388
418
// We might revisit this when https://github.com/apple/swift-nio/issues/970 is resolved.
389
419
connection. channel. eventLoop. execute {
390
- self . execute ( action)
420
+ self . execute ( action, logger : logger )
391
421
}
392
422
}
393
423
}
394
424
395
- func remoteClosed( connection: Connection ) {
425
+ func remoteClosed( connection: Connection , logger : Logger ) {
396
426
let action : Action = self . lock. withLock {
397
427
self . state. remoteClosed ( connection: connection)
398
428
}
399
429
400
- self . execute ( action)
430
+ self . execute ( action, logger : logger )
401
431
}
402
432
403
- func timeout( connection: Connection ) {
433
+ func timeout( connection: Connection , logger : Logger ) {
404
434
let action : Action = self . lock. withLock {
405
435
self . state. timeout ( connection: connection)
406
436
}
407
437
408
- self . execute ( action)
438
+ self . execute ( action, logger : logger )
409
439
}
410
440
411
441
private func closeAndDelete( ) {
@@ -510,25 +540,27 @@ class IdlePoolConnectionHandler: ChannelInboundHandler, RemovableChannelHandler
510
540
511
541
let connection : Connection
512
542
var eventSent : Bool
543
+ let logger : Logger
513
544
514
- init ( connection: Connection ) {
545
+ init ( connection: Connection , logger : Logger ) {
515
546
self . connection = connection
516
547
self . eventSent = false
548
+ self . logger = logger
517
549
}
518
550
519
551
// this is needed to detect when remote end closes connection while connection is in the pool idling
520
552
func channelInactive( context: ChannelHandlerContext ) {
521
553
if !self . eventSent {
522
554
self . eventSent = true
523
- self . connection. remoteClosed ( )
555
+ self . connection. remoteClosed ( logger : self . logger )
524
556
}
525
557
}
526
558
527
559
func userInboundEventTriggered( context: ChannelHandlerContext , event: Any ) {
528
560
if let idleEvent = event as? IdleStateHandler . IdleStateEvent , idleEvent == . write {
529
561
if !self . eventSent {
530
562
self . eventSent = true
531
- self . connection. timeout ( )
563
+ self . connection. timeout ( logger : self . logger )
532
564
}
533
565
} else {
534
566
context. fireUserInboundEventTriggered ( event)
0 commit comments