Add an HTTP/1.1 connection pool (swift-server#105)

motivation: Better performance thanks to connection reuse changes: - Added a connection pool for HTTP/1.1 - All requests automatically use the connection pool - Up to 8 parallel connections per (scheme, host, port) - Multiple additional unit tests

Author: Trevör

Sources/AsyncHTTPClient/ConnectionPool.swift

@@ -69,6 +69,7 @@ internal final class HTTPClientProxyHandler: ChannelDuplexHandler, RemovableChan
 case awaitingResponse
 case connecting
 case connected
+case failed
 }
 
 privatelethost:String
@@ -102,6 +103,7 @@ internal final class HTTPClientProxyHandler: ChannelDuplexHandler, RemovableChan
  // blank line that concludes the successful response's header section
 break
 case407:
+self.readState =.failed
  context.fireErrorCaught(HTTPClientError.proxyAuthenticationRequired)
 default:
  // Any response other than a successful response
@@ -119,6 +121,8 @@ internal final class HTTPClientProxyHandler: ChannelDuplexHandler, RemovableChan
 self.readBuffer.append(data)
 case.connected:
  context.fireChannelRead(data)
+case.failed:
+break
 }
 }
 

Sources/AsyncHTTPClient/HTTPClient.swift

@@ -17,6 +17,7 @@ import NIO
 import NIOConcurrencyHelpers
 import NIOFoundationCompat
 import NIOHTTP1
+import NIOHTTPCompression
 import NIOSSL
 
 extensionHTTPClient{
@@ -486,22 +487,31 @@ extension URL{
 extensionHTTPClient{
  /// Response execution context. Will be created by the library and could be used for obtaining
  /// `EventLoopFuture<Response>` of the execution or cancellation of the execution.
-publicfinalclassTask<Response>{
+publicfinalclassTask<Response>:TaskProtocol{
  /// The `EventLoop` the delegate will be executed on.
 publicleteventLoop:EventLoop
 
 letpromise:EventLoopPromise<Response>
-varchannel:Channel?
-privatevarcancelled:Bool
-privateletlock:Lock
+varcompletion:EventLoopFuture<Void>
+varconnection:ConnectionPool.Connection?
+varcancelled:Bool
+letlock:Lock
+letid=UUID()
 
 init(eventLoop:EventLoop){
 self.eventLoop = eventLoop
 self.promise = eventLoop.makePromise()
+self.completion =self.promise.futureResult.map{ _ in}
 self.cancelled =false
 self.lock =Lock()
 }
 
+staticfunc failedTask(eventLoop:EventLoop, error:Error)->Task<Response>{
+lettask=self.init(eventLoop: eventLoop)
+ task.promise.fail(error)
+return task
+}
+
  /// `EventLoopFuture` for the response returned by this request.
 publicvarfutureResult:EventLoopFuture<Response>{
 returnself.promise.futureResult
@@ -520,28 +530,74 @@ extension HTTPClient{
 letchannel:Channel?=self.lock.withLock{
 if !cancelled {
  cancelled =true
-returnself.channel
+returnself.connection?.channel
+}else{
+returnnil
 }
-returnnil
 }
  channel?.triggerUserOutboundEvent(TaskCancelEvent(), promise:nil)
 }
 
 @discardableResult
-funcsetChannel(_ channel:Channel)->Channel{
+funcsetConnection(_ connection:ConnectionPool.Connection)->ConnectionPool.Connection{
 returnself.lock.withLock{
-self.channel = channel
-return channel
+self.connection = connection
+ifself.cancelled {
+ connection.channel.triggerUserOutboundEvent(TaskCancelEvent(), promise:nil)
+}
+return connection
+}
+}
+
+func succeed<Delegate:HTTPClientResponseDelegate>(promise:EventLoopPromise<Response>?, with value:Response, delegateType:Delegate.Type){
+self.releaseAssociatedConnection(delegateType: delegateType).whenSuccess{
+ promise?.succeed(value)
+}
+}
+
+func fail<Delegate:HTTPClientResponseDelegate>(with error:Error, delegateType:Delegate.Type){
+iflet connection =self.connection {
+ connection.close().whenComplete{ _ in
+self.releaseAssociatedConnection(delegateType: delegateType).whenComplete{ _ in
+self.promise.fail(error)
+}
+}
+}
+}
+
+func releaseAssociatedConnection<Delegate:HTTPClientResponseDelegate>(delegateType:Delegate.Type)->EventLoopFuture<Void>{
+iflet connection =self.connection {
+return connection.removeHandler(NIOHTTPResponseDecompressor.self).flatMap{
+ connection.removeHandler(IdleStateHandler.self)
+}.flatMap{
+ connection.removeHandler(TaskHandler<Delegate>.self)
+}.map{
+ connection.release()
+}.flatMapError{ error in
+fatalError("Couldn't remove taskHandler: \(error)")
+}
+
+}else{
+ // TODO: This seems only reached in some internal unit test
+ // Maybe there could be a better handling in the future to make
+ // it an error outside of testing contexts
+returnself.eventLoop.makeSucceededFuture(())
 }
 }
 }
 }
 
 internalstructTaskCancelEvent{}
 
+internalprotocolTaskProtocol{
+func cancel()
+varid:UUID{get}
+varcompletion:EventLoopFuture<Void>{get}
+}
+
 // MARK: - TaskHandler
 
-internalclassTaskHandler<Delegate:HTTPClientResponseDelegate>{
+internalclassTaskHandler<Delegate:HTTPClientResponseDelegate>:RemovableChannelHandler{
 enumState{
 case idle
 case sent
@@ -581,7 +637,7 @@ extension TaskHandler{
  _ body:@escaping(HTTPClient.Task<Delegate.Response>,Err)->Void){
 func doIt(){
 body(self.task, error)
-self.task.promise.fail(error)
+self.task.fail(with:error, delegateType:Delegate.self)
 }
 
 ifself.task.eventLoop.inEventLoop {
@@ -621,13 +677,14 @@ extension TaskHandler{
 }
 
 func callOutToDelegate<Response>(promise:EventLoopPromise<Response>?=nil,
- _ body:@escaping(HTTPClient.Task<Delegate.Response>)throws->Response){
+ _ body:@escaping(HTTPClient.Task<Delegate.Response>)throws->Response)where Response ==Delegate.Response{
 func doIt(){
 do{
 letresult=trybody(self.task)
- promise?.succeed(result)
+
+self.task.succeed(promise: promise, with: result, delegateType:Delegate.self)
 }catch{
-promise?.fail(error)
+self.task.fail(with:error, delegateType:Delegate.self)
 }
 }
 
@@ -641,7 +698,7 @@ extension TaskHandler{
 }
 
 func callOutToDelegate<Response>(channelEventLoop:EventLoop,
- _ body:@escaping(HTTPClient.Task<Delegate.Response>)throws->Response)->EventLoopFuture<Response>{
+ _ body:@escaping(HTTPClient.Task<Delegate.Response>)throws->Response)->EventLoopFuture<Response>where Response ==Delegate.Response{
 letpromise= channelEventLoop.makePromise(of:Response.self)
 self.callOutToDelegate(promise: promise, body)
 return promise.futureResult
@@ -678,8 +735,6 @@ extension TaskHandler: ChannelDuplexHandler{
  headers.add(name:"Host", value: request.host)
 }
 
- headers.add(name:"Connection", value:"close")
-
 do{
 try headers.validate(body: request.body)
 }catch{
@@ -702,16 +757,10 @@ extension TaskHandler: ChannelDuplexHandler{
  context.eventLoop.assertInEventLoop()
 self.state =.sent
 self.callOutToDelegateFireAndForget(self.delegate.didSendRequest)
-
-letchannel= context.channel
-self.task.futureResult.whenComplete{ _ in
- channel.close(promise:nil)
-}
 }.flatMapErrorThrowing{ error in
  context.eventLoop.assertInEventLoop()
 self.state =.end
 self.failTaskAndNotifyDelegate(error: error,self.delegate.didReceiveError)
- context.close(promise:nil)
 throw error
 }.cascade(to: promise)
 }
@@ -742,6 +791,16 @@ extension TaskHandler: ChannelDuplexHandler{
 letresponse=self.unwrapInboundIn(data)
 switch response {
 case.head(let head):
+if !head.isKeepAlive {
+self.task.lock.withLock{
+iflet connection =self.task.connection {
+ connection.isClosing =true
+}else{
+preconditionFailure("There should always be a connection at this point")
+}
+}
+}
+
 iflet redirectURL = redirectHandler?.redirectTarget(status: head.status, headers: head.headers){
 self.state =.redirected(head, redirectURL)
 }else{
@@ -768,8 +827,9 @@ extension TaskHandler: ChannelDuplexHandler{
 switchself.state {
 case.redirected(let head,let redirectURL):
 self.state =.end
-self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise:self.task.promise)
- context.close(promise:nil)
+self.task.releaseAssociatedConnection(delegateType:Delegate.self).whenSuccess{
+self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise:self.task.promise)
+}
 default:
 self.state =.end
 self.callOutToDelegate(promise:self.task.promise,self.delegate.didFinishRequest)
@@ -845,6 +905,13 @@ extension TaskHandler: ChannelDuplexHandler{
 self.failTaskAndNotifyDelegate(error: error,self.delegate.didReceiveError)
 }
 }
+
+func handlerAdded(context:ChannelHandlerContext){
+guard context.channel.isActive else{
+self.failTaskAndNotifyDelegate(error:HTTPClientError.remoteConnectionClosed,self.delegate.didReceiveError)
+return
+}
+}
 }
 
 // MARK: - RedirectHandler
@@ -931,9 +998,13 @@ internal struct RedirectHandler<ResponseType>{
 do{
 varnewRequest=tryHTTPClient.Request(url: redirectURL, method: method, headers: headers, body: body)
  newRequest.redirectState = nextState
-returnself.execute(newRequest).futureResult.cascade(to: promise)
+self.execute(newRequest).futureResult.whenComplete{ result in
+ promise.futureResult.eventLoop.execute{
+ promise.completeWith(result)
+}
+}
 }catch{
-returnpromise.fail(error)
+ promise.fail(error)
 }
 }
 }

Sources/AsyncHTTPClient/HTTPClientProxyHandler.swift

@@ -14,6 +14,7 @@
 
 import NIO
 import NIOHTTP1
+import NIOHTTPCompression
 
 internalextensionString{
 varisIPAddress:Bool{
@@ -44,3 +45,53 @@ public final class HTTPClientCopyingDelegate: HTTPClientResponseDelegate{
 return()
 }
 }
+
+extensionClientBootstrap{
+staticfunc makeHTTPClientBootstrapBase(group:EventLoopGroup, host:String, port:Int, configuration:HTTPClient.Configuration, channelInitializer:((Channel)->EventLoopFuture<Void>)?=nil)->ClientBootstrap{
+returnClientBootstrap(group: group)
+.channelOption(ChannelOptions.socket(SocketOptionLevel(IPPROTO_TCP), TCP_NODELAY), value:1)
+
+.channelInitializer{ channel in
+letchannelAddedFuture:EventLoopFuture<Void>
+switch configuration.proxy {
+case.none:
+ channelAddedFuture = group.next().makeSucceededFuture(())
+case.some:
+ channelAddedFuture = channel.pipeline.addProxyHandler(host: host, port: port, authorization: configuration.proxy?.authorization)
+}
+return channelAddedFuture.flatMap{(_:Void)->EventLoopFuture<Void>in
+channelInitializer?(channel)?? group.next().makeSucceededFuture(())
+}
+}
+}
+}
+
+extensionCircularBuffer{
+@discardableResult
+mutatingfunc swapWithFirstAndRemove(at index:Index)->Element?{
+precondition(index >=self.startIndex && index <self.endIndex)
+if !self.isEmpty {
+self.swapAt(self.startIndex, index)
+returnself.removeFirst()
+}else{
+returnnil
+}
+}
+
+@discardableResult
+mutatingfunc swapWithFirstAndRemove(where predicate:(Element)throws->Bool)rethrows->Element?{
+iflet existingIndex =tryself.firstIndex(where: predicate){
+returnself.swapWithFirstAndRemove(at: existingIndex)
+}else{
+returnnil
+}
+}
+}
+
+extensionConnectionPool.Connection{
+func removeHandler<Handler:RemovableChannelHandler>(_ type:Handler.Type)->EventLoopFuture<Void>{
+returnself.channel.pipeline.handler(type: type).flatMap{ handler in
+self.channel.pipeline.removeHandler(handler)
+}.recover{ _ in}
+}
+}

Sources/AsyncHTTPClient/HTTPHandler.swift

@@ -33,6 +33,7 @@ extension HTTPClientInternalTests{
 ("testUploadStreamingBackpressure", testUploadStreamingBackpressure),
 ("testRequestURITrailingSlash", testRequestURITrailingSlash),
 ("testChannelAndDelegateOnDifferentEventLoops", testChannelAndDelegateOnDifferentEventLoops),
+("testResponseConnectionCloseGet", testResponseConnectionCloseGet),
 ]
 }
 }