Make HTTP/2 flow control target window size configurable in a server connection.

Motivation:

Higher streaming thruput from client to server is available by specifying a flow control window size higher than the initial value of 2^16-1. The target value needs to be passed either using SETTINGS_INITIAL_WINDOW_SIZE in the connection preface, or via a WINDOW_UPDATE. The HTTP2FlowControlWindow passes the value configured here in `windowUpdate`.

Modifications:
Added property `httpTargetWindowSize` to `Server.Configuration`.
In `Server.Configuration.init`, added httpTargetWindowSize arg with default value 2^16-1.
Added property and setter for `httpTargetWindowSize` to `Server.Builder`.
During server bootstrap in Server.start, pass `configuration.httpTargetWindowSize` to the HTTPProtocolSwitcher.
Configure http/2 flow control target window size on the http2 pipeline in HTTPProxySwitcher.

Result:

One can build a `Server` with a target window size value that is configured via a WINDOW_UPDATE after the connection is established.

Author: John Kassebaum

Sources/GRPC/HTTPProtocolSwitcher.swift

@@ -25,6 +25,7 @@ internal class HTTPProtocolSwitcher {
   private let handlersInitializer: ((Channel) -> EventLoopFuture<Void>)
   private let errorDelegate: ServerErrorDelegate?
   private let logger = Logger(subsystem: .serverChannelCall)
+  private let httpTargetWindowSize: Int
 
   // We could receive additional data after the initial data and before configuring
   // the pipeline; buffer it and fire it down the pipeline once it is configured.
@@ -41,8 +42,13 @@ internal class HTTPProtocolSwitcher {
   }
   private var bufferedData: [NIOAny] = []
 
-  init(errorDelegate: ServerErrorDelegate?, handlersInitializer: (@escaping (Channel) -> EventLoopFuture<Void>)) {
+  init(
+    errorDelegate: ServerErrorDelegate?,
+    httpTargetWindowSize: Int = 65535,
+    handlersInitializer: (@escaping (Channel) -> EventLoopFuture<Void>)
+  ) {
     self.errorDelegate = errorDelegate
+    self.httpTargetWindowSize = httpTargetWindowSize
     self.handlersInitializer = handlersInitializer
   }
 }
@@ -132,7 +138,10 @@ extension HTTPProtocolSwitcher: ChannelInboundHandler, RemovableChannelHandler {
           .cascade(to: pipelineConfigured)
 
       case .http2:
-        context.channel.configureHTTP2Pipeline(mode: .server) { (streamChannel, streamID) in
+        context.channel.configureHTTP2Pipeline(
+          mode: .server,
+          targetWindowSize: httpTargetWindowSize
+        ) { (streamChannel, streamID) in
             streamChannel.pipeline.addHandler(HTTP2ToHTTP1ServerCodec(streamID: streamID, normalizeHTTPHeaders: true))
               .flatMap { self.handlersInitializer(streamChannel) }
           }

Sources/GRPC/Server.swift

@@ -99,7 +99,10 @@ public final class Server {
       .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
       // Set the handlers that are applied to the accepted Channels
       .childChannelInitializer { channel in
-        let protocolSwitcher = HTTPProtocolSwitcher(errorDelegate: configuration.errorDelegate) { channel -> EventLoopFuture<Void> in
+        let protocolSwitcher = HTTPProtocolSwitcher(
+          errorDelegate: configuration.errorDelegate,
+          httpTargetWindowSize: configuration.httpTargetWindowSize
+        ) { channel -> EventLoopFuture<Void> in
           let logger = Logger(subsystem: .serverChannelCall, metadata: [MetadataKey.requestID: "\(UUID())"])
           let handler = GRPCServerRequestRoutingHandler(
             servicesByName: configuration.serviceProvidersByName,
@@ -173,7 +176,7 @@ extension Server {
   public struct Configuration {
     /// The target to bind to.
     public var target: BindTarget
-
+    
     /// The event loop group to run the connection on.
     public var eventLoopGroup: EventLoopGroup
 
@@ -196,6 +199,9 @@ extension Server {
     /// streaming and bidirectional streaming RPCs) by passing setting `compression` to `.disabled`
     /// in `sendResponse(_:compression)`.
     public var messageEncoding: ServerMessageEncoding
+    
+    /// The HTTP/2 flow control target window size.
+    public var httpTargetWindowSize: Int
 
     /// Create a `Configuration` with some pre-defined defaults.
     ///
@@ -212,14 +218,16 @@ extension Server {
       serviceProviders: [CallHandlerProvider],
       errorDelegate: ServerErrorDelegate? = LoggingServerErrorDelegate.shared,
       tls: TLS? = nil,
-      messageEncoding: ServerMessageEncoding = .disabled
+      messageEncoding: ServerMessageEncoding = .disabled,
+      httpTargetWindowSize: Int = 65535
     ) {
       self.target = target
       self.eventLoopGroup = eventLoopGroup
       self.serviceProviders = serviceProviders
       self.errorDelegate = errorDelegate
       self.tls = tls
       self.messageEncoding = messageEncoding
+      self.httpTargetWindowSize = httpTargetWindowSize
     }
   }
 }

Sources/GRPC/ServerBuilder.swift

@@ -23,6 +23,7 @@ extension Server {
     private var providers: [CallHandlerProvider] = []
     private var errorDelegate: ServerErrorDelegate?
     private var messageEncoding: ServerMessageEncoding = .disabled
+    private var httpTargetWindowSize: Int = 65535
 
     fileprivate init(group: EventLoopGroup) {
       self.group = group
@@ -50,7 +51,8 @@ extension Server {
         serviceProviders: self.providers,
         errorDelegate: self.errorDelegate,
         tls: self.maybeTLS,
-        messageEncoding: self.messageEncoding
+        messageEncoding: self.messageEncoding,
+        httpTargetWindowSize: self.httpTargetWindowSize
       )
       return Server.start(configuration: configuration)
     }
@@ -102,6 +104,15 @@ extension Server.Builder.Secure {
   }
 }
 
+extension Server.Builder {
+  /// Sets the HTTP/2 flow control target window size.
+  @discardableResult
+  public func withHttpTargetWindowSize(_ httpTargetWindowSize: Int) -> Self {
+    self.httpTargetWindowSize = httpTargetWindowSize
+    return self
+  }
+}
+
 extension Server {
   /// Returns an insecure `Server` builder which is *not configured with TLS*.
   public static func insecure(group: EventLoopGroup) -> Builder {