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

Motivation:

Higher stremaing thruput from server to client 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 `targetWindowSize` to `ClientConnection.Configuration`.
In the existing `ClientConnection.Configuration.init`, sets the new `targetWindowSize` property to the initial flow-control window size (65535).
Duplicated `ClientConnection.Configuration.init` and added a `targetWindowSize` arg.
Added property and setter for `targetWindowSize` to `ClientConnection.Builder`.
During client bootstrap, pass `configuration.targetWindowSize` to the channel initializer.
Duplicated channel initializer `configureGRPCClient` and added a `targetWindowSize` arg.
Duplicated all tests that directly invoke the original `ClientConnection.Configuration.init` to explicitly pass `targetWindowSize`to the new initializer.

Result:

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

Author: John Kassebaum

Sources/GRPC/ClientConnection.swift

@@ -419,6 +419,7 @@ extension ClientConnection {
       .channelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
       .channelInitializer { channel in
         channel.configureGRPCClient(
+          targetWindowSize: configuration.targetWindowSize,
           tlsConfiguration: configuration.tls?.configuration,
           tlsServerHostname: serverHostname,
           connectivityMonitor: connectivityMonitor,
@@ -467,6 +468,9 @@ extension ClientConnection {
   public struct Configuration {
     /// The target to connect to.
     public var target: ConnectionTarget
+    
+    ///The HTTP/2 flow control target window size.
+    public var targetWindowSize: Int
 
     /// The event loop group to run the connection on.
     public var eventLoopGroup: EventLoopGroup
@@ -512,6 +516,38 @@ extension ClientConnection {
       connectionBackoff: ConnectionBackoff? = ConnectionBackoff()
     ) {
       self.target = target
+      self.targetWindowSize = 65535
+      self.eventLoopGroup = eventLoopGroup
+      self.errorDelegate = errorDelegate
+      self.connectivityStateDelegate = connectivityStateDelegate
+      self.tls = tls
+      self.connectionBackoff = connectionBackoff
+    }
+    
+    /// Create a `Configuration` with some pre-defined defaults. Prefer using
+    /// `ClientConnection.secure(group:)` to build a connection secured with TLS or
+    /// `ClientConnection.insecure(group:)` to build a plaintext connection.
+    ///
+    /// - Parameter target: The target to connect to.
+    /// - Parameter targetWindowSize: The HTTP/2 flow control target window size.
+    /// - Parameter eventLoopGroup: The event loop group to run the connection on.
+    /// - Parameter errorDelegate: The error delegate, defaulting to a delegate which will log only
+    ///     on debug builds.
+    /// - Parameter connectivityStateDelegate: A connectivity state delegate, defaulting to `nil`.
+    /// - Parameter tlsConfiguration: TLS configuration, defaulting to `nil`.
+    /// - Parameter connectionBackoff: The connection backoff configuration to use.
+    /// - Parameter messageEncoding: Message compression configuration, defaults to no compression.
+    public init(
+      target: ConnectionTarget,
+      targetWindowSize: Int,
+      eventLoopGroup: EventLoopGroup,
+      errorDelegate: ClientErrorDelegate? = LoggingClientErrorDelegate(),
+      connectivityStateDelegate: ConnectivityStateDelegate? = nil,
+      tls: Configuration.TLS? = nil,
+      connectionBackoff: ConnectionBackoff? = ConnectionBackoff()
+    ) {
+      self.target = target
+      self.targetWindowSize = targetWindowSize
       self.eventLoopGroup = eventLoopGroup
       self.errorDelegate = errorDelegate
       self.connectivityStateDelegate = connectivityStateDelegate
@@ -601,6 +637,33 @@ extension Channel {
     }
   }
 
+  func configureGRPCClient(
+    targetWindowSize: Int,
+    tlsConfiguration: TLSConfiguration?,
+    tlsServerHostname: String?,
+    connectivityMonitor: ConnectivityStateMonitor,
+    errorDelegate: ClientErrorDelegate?,
+    logger: Logger
+  ) -> EventLoopFuture<Void> {
+    let tlsConfigured = tlsConfiguration.map {
+      self.configureTLS($0, serverHostname: tlsServerHostname, errorDelegate: errorDelegate, logger: logger)
+    }
+
+    return (tlsConfigured ?? self.eventLoop.makeSucceededFuture(())).flatMap {
+      self.configureHTTP2Pipeline(mode: .client, targetWindowSize: targetWindowSize)
+    }.flatMap { _ in
+      let settingsObserver = InitialSettingsObservingHandler(
+        connectivityStateMonitor: connectivityMonitor,
+        logger: logger
+      )
+      let errorHandler = DelegatingErrorHandler(
+        logger: logger,
+        delegate: errorDelegate
+      )
+      return self.pipeline.addHandlers(settingsObserver, errorHandler)
+    }
+  }
+
   func configureGRPCClient(
     errorDelegate: ClientErrorDelegate?,
     logger: Logger

Sources/GRPC/GRPCChannel/GRPCChannelBuilder.swift

@@ -31,6 +31,7 @@ extension ClientConnection {
 extension ClientConnection {
   public class Builder {
     private let group: EventLoopGroup
+    private var targetWindowSize: Int = 65535
     private var maybeTLS: ClientConnection.Configuration.TLS? { return nil }
     private var connectionBackoff = ConnectionBackoff()
     private var connectionBackoffIsEnabled = true
@@ -44,6 +45,7 @@ extension ClientConnection {
     public func connect(host: String, port: Int) -> ClientConnection {
       let configuration = ClientConnection.Configuration(
         target: .hostAndPort(host, port),
+        targetWindowSize: self.targetWindowSize,
         eventLoopGroup: self.group,
         errorDelegate: self.errorDelegate,
         connectivityStateDelegate: self.connectivityStateDelegate,
@@ -132,6 +134,15 @@ extension ClientConnection.Builder {
   }
 }
 
+extension ClientConnection.Builder {
+  /// Sets the HTTP/2 flow control target window size.
+  @discardableResult
+  public func withTargetWindowSize(_ targetWindowSize: Int) -> Self {
+    self.targetWindowSize = targetWindowSize
+    return self
+  }
+}
+
 extension ClientConnection.Builder {
   /// Sets the client error delegate.
   @discardableResult

Tests/GRPCTests/ClientTLSFailureTests.swift

@@ -65,6 +65,20 @@ class ClientTLSFailureTests: GRPCTestCase {
     )
   }
 
+  func makeClientConfiguration(
+    tls: ClientConnection.Configuration.TLS,
+    targetWindowSize: Int
+  ) -> ClientConnection.Configuration {
+    return .init(
+      target: .hostAndPort("localhost", self.port),
+      targetWindowSize: targetWindowSize,
+      eventLoopGroup: self.clientEventLoopGroup,
+      tls: tls,
+      // No need to retry connecting.
+      connectionBackoff: nil
+    )
+  }
+  
   func makeClientConnectionExpectation() -> XCTestExpectation {
     return self.expectation(description: "EventLoopFuture<ClientConnection> resolved")
   }
@@ -123,6 +137,32 @@ class ClientTLSFailureTests: GRPCTestCase {
       XCTFail("Expected NIOSSLError.handshakeFailed(BoringSSL.sslError)")
     }
   }
+  
+  func testClientConnectionFailsWhenServerIsUnknownWithTargetWindowSize() throws {
+    let shutdownExpectation = self.expectation(description: "client shutdown")
+    let errorExpectation = self.expectation(description: "error")
+
+    var tls = self.defaultClientTLSConfiguration
+    tls.trustRoots = .certificates([])
+    var configuration = self.makeClientConfiguration(tls: tls, targetWindowSize: 65535)
+
+    let errorRecorder = ErrorRecordingDelegate(expectation: errorExpectation)
+    configuration.errorDelegate = errorRecorder
+
+    let stateChangeDelegate = ConnectivityStateCollectionDelegate(shutdown: shutdownExpectation)
+    configuration.connectivityStateDelegate = stateChangeDelegate
+
+    _ = ClientConnection(configuration: configuration)
+
+    self.wait(for: [shutdownExpectation, errorExpectation], timeout: self.defaultTestTimeout)
+
+    if let nioSSLError = errorRecorder.errors.first as? NIOSSLError,
+      case .handshakeFailed(.sslError) = nioSSLError {
+      // Expected case.
+    } else {
+      XCTFail("Expected NIOSSLError.handshakeFailed(BoringSSL.sslError)")
+    }
+  }
 
   func testClientConnectionFailsWhenHostnameIsNotValid() throws {
     let shutdownExpectation = self.expectation(description: "client shutdown")
@@ -149,4 +189,30 @@ class ClientTLSFailureTests: GRPCTestCase {
       XCTFail("Expected NIOSSLExtraError.failedToValidateHostname")
     }
   }
+  
+  func testClientConnectionFailsWhenHostnameIsNotValidWithTargetWindowSize() throws {
+    let shutdownExpectation = self.expectation(description: "client shutdown")
+    let errorExpectation = self.expectation(description: "error")
+
+    var tls = self.defaultClientTLSConfiguration
+    tls.hostnameOverride = "not-the-server-hostname"
+
+    var configuration = self.makeClientConfiguration(tls: tls, targetWindowSize: 65535)
+    let errorRecorder = ErrorRecordingDelegate(expectation: errorExpectation)
+    configuration.errorDelegate = errorRecorder
+
+    let stateChangeDelegate = ConnectivityStateCollectionDelegate(shutdown: shutdownExpectation)
+    configuration.connectivityStateDelegate = stateChangeDelegate
+
+    let _ = ClientConnection(configuration: configuration)
+
+    self.wait(for: [shutdownExpectation, errorExpectation], timeout: self.defaultTestTimeout)
+
+    if let nioSSLError = errorRecorder.errors.first as? NIOSSLExtraError {
+      XCTAssertEqual(nioSSLError, .failedToValidateHostname)
+      // Expected case.
+    } else {
+      XCTFail("Expected NIOSSLExtraError.failedToValidateHostname")
+    }
+  }
 }