Revert "Adopt h2handler multiplexer (#1587)" (#1641)

This reverts commit 75b390e.

.github/workflows/ci.yaml

@@ -57,45 +57,44 @@ jobs:
         include:
           - image: swiftlang/swift:nightly-focal
             env:
-              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 246000
-              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_1_request: 138000
+              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 323000
+              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_1_request: 161000
               MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_10_small_requests: 110000
               MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_1_small_request: 65000
               MAX_ALLOCS_ALLOWED_embedded_server_unary_1k_rpcs_1_small_request: 61000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 129000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 136000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 136000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 163000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 170000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 170000
           - image: swift:5.8-jammy
             env:
-              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 246000
-              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_1_request: 138000
+              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 323000
+              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_1_request: 161000
               MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_10_small_requests: 110000
               MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_1_small_request: 65000
               MAX_ALLOCS_ALLOWED_embedded_server_unary_1k_rpcs_1_small_request: 61000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 129000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 136000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 136000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 163000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 170000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 170000
           - image: swift:5.7-jammy
             env:
-              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 246000
-              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_1_request: 138000
+              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 323000
+              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_1_request: 161000
               MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_10_small_requests: 110000
               MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_1_small_request: 65000
               MAX_ALLOCS_ALLOWED_embedded_server_unary_1k_rpcs_1_small_request: 61000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 129000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 136000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 136000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 163000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 170000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 170000
           - image: swift:5.6-focal
             env:
-              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 247000
-              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_1_request: 139000
+              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 324000
+              MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_1_request: 162000
               MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_10_small_requests: 110000
               MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_1_small_request: 65000
               MAX_ALLOCS_ALLOWED_embedded_server_unary_1k_rpcs_1_small_request: 61000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 130000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 137000
-              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 137000
-
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 164000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 171000
+              MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 171000
     name: Performance Tests on ${{ matrix.image }}
     runs-on: ubuntu-latest
     container:

NOTICES.txt

@@ -25,11 +25,3 @@ framework: 'test_01_allocation_counts.sh', 'run-nio-alloc-counter-tests.sh' and
     * https://github.com/apple/swift-nio/blob/main/LICENSE.txt
   * HOMEPAGE:
     * https://github.com/apple/swift-nio
-
-This product contains a simplified derivation of SwiftNIO HTTP/2's
-'HTTP2FrameEncoder' for testing purposes.
-
-  * LICENSE (Apache License 2.0):
-    * https://github.com/apple/swift-nio-http2/blob/main/LICENSE.txt
-  * HOMEPAGE:
-    * https://github.com/apple/swift-nio-http2

Package.swift

@@ -36,7 +36,7 @@ let packageDependencies: [Package.Dependency] = [
   ),
   .package(
     url: "https://github.com/apple/swift-nio-http2.git",
-    from: "1.26.0"
+    from: "1.24.1"
   ),
   .package(
     url: "https://github.com/apple/swift-nio-transport-services.git",

Sources/GRPC/ClientConnection.swift

@@ -58,25 +58,28 @@ import SwiftProtobuf
 ///               │  DelegatingErrorHandler  │
 ///               └──────────▲───────────────┘
 ///                HTTP2Frame│
-///                          │
-///                          │
-///                          │
-///                          │
-///                          │
-///                HTTP2Frame│                  ⠇        ⠇ ⠇   ⠇ ⠇
-///                        ┌─┴──────────────────▼─┐     ┌┴─▼┐ ┌┴─▼┐
-///                        │    GRPCIdleHandler   │     │   | │   | HTTP/2 streams
-///                        └─▲──────────────────┬─┘     └▲─┬┘ └▲─┬┘
-///                HTTP2Frame│                  │        │ │   │ │ HTTP2Frame
-///                        ┌─┴──────────────────▼────────┴─▼───┴─▼┐
-///                        │            NIOHTTP2Handler           │
-///                        └─▲──────────────────────────────────┬─┘
-///                ByteBuffer│                                  │ByteBuffer
-///                        ┌─┴──────────────────────────────────▼─┐
-///                        │             NIOSSLHandler            │
-///                        └─▲──────────────────────────────────┬─┘
-///                ByteBuffer│                                  │ByteBuffer
-///                          │                                  ▼
+///                          │                ⠇ ⠇   ⠇ ⠇
+///                          │               ┌┴─▼┐ ┌┴─▼┐
+///                          │               │   | │   | HTTP/2 streams
+///                          │               └▲─┬┘ └▲─┬┘
+///                          │                │ │   │ │ HTTP2Frame
+///                        ┌─┴────────────────┴─▼───┴─▼┐
+///                        │   HTTP2StreamMultiplexer  |
+///                        └─▲───────────────────────┬─┘
+///                HTTP2Frame│                       │HTTP2Frame
+///                        ┌─┴───────────────────────▼─┐
+///                        │       GRPCIdleHandler     │
+///                        └─▲───────────────────────┬─┘
+///                HTTP2Frame│                       │HTTP2Frame
+///                        ┌─┴───────────────────────▼─┐
+///                        │       NIOHTTP2Handler     │
+///                        └─▲───────────────────────┬─┘
+///                ByteBuffer│                       │ByteBuffer
+///                        ┌─┴───────────────────────▼─┐
+///                        │       NIOSSLHandler       │
+///                        └─▲───────────────────────┬─┘
+///                ByteBuffer│                       │ByteBuffer
+///                          │                       ▼
 ///
 /// The 'GRPCIdleHandler' intercepts HTTP/2 frames and various events and is responsible for
 /// informing and controlling the state of the connection (idling and keepalive). The HTTP/2 streams
@@ -85,7 +88,7 @@ public final class ClientConnection: Sendable {
   private let connectionManager: ConnectionManager
 
   /// HTTP multiplexer from the underlying channel handling gRPC calls.
-  internal func getMultiplexer() -> EventLoopFuture<NIOHTTP2Handler.StreamMultiplexer> {
+  internal func getMultiplexer() -> EventLoopFuture<HTTP2StreamMultiplexer> {
     return self.connectionManager.getHTTP2Multiplexer()
   }
 
@@ -247,7 +250,7 @@ extension ClientConnection: GRPCChannel {
   }
 
   private static func makeStreamChannel(
-    using result: Result<NIOHTTP2Handler.StreamMultiplexer, Error>,
+    using result: Result<HTTP2StreamMultiplexer, Error>,
     promise: EventLoopPromise<Channel>
   ) {
     switch result {
@@ -618,31 +621,29 @@ extension ChannelPipeline.SynchronousOperations {
       HTTP2Setting(parameter: .initialWindowSize, value: httpTargetWindowSize),
     ]
 
-    let grpcIdleHandler = GRPCIdleHandler(
+    // We could use 'configureHTTP2Pipeline' here, but we need to add a few handlers between the
+    // two HTTP/2 handlers so we'll do it manually instead.
+    try self.addHandler(NIOHTTP2Handler(mode: .client, initialSettings: initialSettings))
+
+    let h2Multiplexer = HTTP2StreamMultiplexer(
+      mode: .client,
+      channel: channel,
+      targetWindowSize: httpTargetWindowSize,
+      inboundStreamInitializer: nil
+    )
+
+    // The multiplexer is passed through the idle handler so it is only reported on
+    // successful channel activation - with happy eyeballs multiple pipelines can
+    // be constructed so it's not safe to report just yet.
+    try self.addHandler(GRPCIdleHandler(
       connectionManager: connectionManager,
+      multiplexer: h2Multiplexer,
       idleTimeout: connectionIdleTimeout,
       keepalive: connectionKeepalive,
       logger: logger
-    )
-
-    var connectionConfiguration = NIOHTTP2Handler.ConnectionConfiguration()
-    connectionConfiguration.initialSettings = initialSettings
-    var streamConfiguration = NIOHTTP2Handler.StreamConfiguration()
-    streamConfiguration.targetWindowSize = httpTargetWindowSize
-    let h2Handler = NIOHTTP2Handler(
-      mode: .client,
-      eventLoop: channel.eventLoop,
-      connectionConfiguration: connectionConfiguration,
-      streamConfiguration: streamConfiguration,
-      streamDelegate: grpcIdleHandler
-    ) { channel in
-      channel.close()
-    }
-    try self.addHandler(h2Handler)
-
-    grpcIdleHandler.setMultiplexer(try h2Handler.syncMultiplexer())
-    try self.addHandler(grpcIdleHandler)
+    ))
 
+    try self.addHandler(h2Multiplexer)
     try self.addHandler(DelegatingErrorHandler(logger: logger, delegate: errorDelegate))
   }
 }
@@ -652,13 +653,7 @@ extension Channel {
     errorDelegate: ClientErrorDelegate?,
     logger: Logger
   ) -> EventLoopFuture<Void> {
-    return self.configureHTTP2Pipeline(
-      mode: .client,
-      connectionConfiguration: .init(),
-      streamConfiguration: .init()
-    ) { channel in
-      channel.eventLoop.makeSucceededVoidFuture()
-    }.flatMap { _ in
+    return self.configureHTTP2Pipeline(mode: .client, inboundStreamInitializer: nil).flatMap { _ in
       self.pipeline.addHandler(DelegatingErrorHandler(logger: logger, delegate: errorDelegate))
     }
   }

Sources/GRPC/ConnectionManager.swift

@@ -35,23 +35,19 @@ internal final class ConnectionManager: @unchecked Sendable {
     var reconnect: Reconnect
 
     var candidate: EventLoopFuture<Channel>
-    var readyChannelMuxPromise: EventLoopPromise<NIOHTTP2Handler.StreamMultiplexer>
-    var candidateMuxPromise: EventLoopPromise<NIOHTTP2Handler.StreamMultiplexer>
+    var readyChannelMuxPromise: EventLoopPromise<HTTP2StreamMultiplexer>
+    var candidateMuxPromise: EventLoopPromise<HTTP2StreamMultiplexer>
   }
 
   internal struct ConnectedState {
     var backoffIterator: ConnectionBackoffIterator?
     var reconnect: Reconnect
     var candidate: Channel
-    var readyChannelMuxPromise: EventLoopPromise<NIOHTTP2Handler.StreamMultiplexer>
-    var multiplexer: NIOHTTP2Handler.StreamMultiplexer
+    var readyChannelMuxPromise: EventLoopPromise<HTTP2StreamMultiplexer>
+    var multiplexer: HTTP2StreamMultiplexer
     var error: Error?
 
-    init(
-      from state: ConnectingState,
-      candidate: Channel,
-      multiplexer: NIOHTTP2Handler.StreamMultiplexer
-    ) {
+    init(from state: ConnectingState, candidate: Channel, multiplexer: HTTP2StreamMultiplexer) {
       self.backoffIterator = state.backoffIterator
       self.reconnect = state.reconnect
       self.candidate = candidate
@@ -62,7 +58,7 @@ internal final class ConnectionManager: @unchecked Sendable {
 
   internal struct ReadyState {
     var channel: Channel
-    var multiplexer: NIOHTTP2Handler.StreamMultiplexer
+    var multiplexer: HTTP2StreamMultiplexer
     var error: Error?
 
     init(from state: ConnectedState) {
@@ -73,7 +69,7 @@ internal final class ConnectionManager: @unchecked Sendable {
 
   internal struct TransientFailureState {
     var backoffIterator: ConnectionBackoffIterator?
-    var readyChannelMuxPromise: EventLoopPromise<NIOHTTP2Handler.StreamMultiplexer>
+    var readyChannelMuxPromise: EventLoopPromise<HTTP2StreamMultiplexer>
     var scheduled: Scheduled<Void>
     var reason: Error
 
@@ -256,8 +252,8 @@ internal final class ConnectionManager: @unchecked Sendable {
     }
   }
 
-  /// Returns the `NIOHTTP2Handler.StreamMultiplexer` from the 'ready' state or `nil` if it is not available.
-  private var multiplexer: NIOHTTP2Handler.StreamMultiplexer? {
+  /// Returns the `HTTP2StreamMultiplexer` from the 'ready' state or `nil` if it is not available.
+  private var multiplexer: HTTP2StreamMultiplexer? {
     self.eventLoop.assertInEventLoop()
     switch self.state {
     case let .ready(state):
@@ -365,8 +361,8 @@ internal final class ConnectionManager: @unchecked Sendable {
   /// Get the multiplexer from the underlying channel handling gRPC calls.
   /// if the `ConnectionManager` was configured to be `fastFailure` this will have
   /// one chance to connect - if not reconnections are managed here.
-  internal func getHTTP2Multiplexer() -> EventLoopFuture<NIOHTTP2Handler.StreamMultiplexer> {
-    func getHTTP2Multiplexer0() -> EventLoopFuture<NIOHTTP2Handler.StreamMultiplexer> {
+  internal func getHTTP2Multiplexer() -> EventLoopFuture<HTTP2StreamMultiplexer> {
+    func getHTTP2Multiplexer0() -> EventLoopFuture<HTTP2StreamMultiplexer> {
       switch self.callStartBehavior {
       case .waitsForConnectivity:
         return self.getHTTP2MultiplexerPatient()
@@ -386,8 +382,8 @@ internal final class ConnectionManager: @unchecked Sendable {
 
   /// Returns a future for the multiplexer which succeeded when the channel is connected.
   /// Reconnects are handled if necessary.
-  private func getHTTP2MultiplexerPatient() -> EventLoopFuture<NIOHTTP2Handler.StreamMultiplexer> {
-    let multiplexer: EventLoopFuture<NIOHTTP2Handler.StreamMultiplexer>
+  private func getHTTP2MultiplexerPatient() -> EventLoopFuture<HTTP2StreamMultiplexer> {
+    let multiplexer: EventLoopFuture<HTTP2StreamMultiplexer>
 
     switch self.state {
     case .idle:
@@ -425,12 +421,11 @@ internal final class ConnectionManager: @unchecked Sendable {
   /// attempt, or if the state is 'idle' returns the future for the next connection attempt.
   ///
   /// Note: if the state is 'transientFailure' or 'shutdown' then a failed future will be returned.
-  private func getHTTP2MultiplexerOptimistic()
-    -> EventLoopFuture<NIOHTTP2Handler.StreamMultiplexer> {
+  private func getHTTP2MultiplexerOptimistic() -> EventLoopFuture<HTTP2StreamMultiplexer> {
     // `getHTTP2Multiplexer` makes sure we're on the event loop but let's just be sure.
     self.eventLoop.preconditionInEventLoop()
 
-    let muxFuture: EventLoopFuture<NIOHTTP2Handler.StreamMultiplexer> = { () in
+    let muxFuture: EventLoopFuture<HTTP2StreamMultiplexer> = { () in
       switch self.state {
       case .idle:
         self.startConnecting()
@@ -661,7 +656,7 @@ internal final class ConnectionManager: @unchecked Sendable {
   }
 
   /// The connecting channel became `active`. Must be called on the `EventLoop`.
-  internal func channelActive(channel: Channel, multiplexer: NIOHTTP2Handler.StreamMultiplexer) {
+  internal func channelActive(channel: Channel, multiplexer: HTTP2StreamMultiplexer) {
     self.eventLoop.preconditionInEventLoop()
     self.logger.debug("activating connection", metadata: [
       "connectivity_state": "\(self.state.label)",
@@ -978,7 +973,7 @@ extension ConnectionManager {
 
   private func startConnecting(
     backoffIterator: ConnectionBackoffIterator?,
-    muxPromise: EventLoopPromise<NIOHTTP2Handler.StreamMultiplexer>
+    muxPromise: EventLoopPromise<HTTP2StreamMultiplexer>
   ) {
     let timeoutAndBackoff = backoffIterator?.next()
 
@@ -1065,7 +1060,7 @@ extension ConnectionManager {
 
     /// Returns the `multiplexer` from a connection in the `ready` state or `nil` if it is any
     /// other state.
-    internal var multiplexer: NIOHTTP2Handler.StreamMultiplexer? {
+    internal var multiplexer: HTTP2StreamMultiplexer? {
       return self.manager.multiplexer
     }
 

Sources/GRPC/ConnectionPool/ConnectionPool+PerConnectionState.swift

@@ -58,7 +58,7 @@ extension ConnectionPool {
       }
 
       @usableFromInline
-      var multiplexer: NIOHTTP2Handler.StreamMultiplexer
+      var multiplexer: HTTP2StreamMultiplexer
       /// Maximum number of available streams.
       @usableFromInline
       var maxAvailable: Int
@@ -78,7 +78,7 @@ extension ConnectionPool {
 
       /// Increment the reserved streams and return the multiplexer.
       @usableFromInline
-      mutating func reserve() -> NIOHTTP2Handler.StreamMultiplexer {
+      mutating func reserve() -> HTTP2StreamMultiplexer {
         assert(!self.isQuiescing)
         self.reserved += 1
         return self.multiplexer
@@ -132,7 +132,7 @@ extension ConnectionPool {
     ///
     /// The result may be safely unwrapped if `self.availableStreams > 0` when reserving a stream.
     @usableFromInline
-    internal mutating func reserveStream() -> NIOHTTP2Handler.StreamMultiplexer? {
+    internal mutating func reserveStream() -> HTTP2StreamMultiplexer? {
       return self._availability?.reserve()
     }
 

Sources/GRPC/ConnectionPool/ConnectionPool+Waiter.swift

@@ -30,7 +30,7 @@ extension ConnectionPool {
 
     /// The channel initializer.
     @usableFromInline
-    internal let _channelInitializer: @Sendable (Channel) -> EventLoopFuture<Void>
+    internal let _channelInitializer: (Channel) -> EventLoopFuture<Void>
 
     /// The deadline at which the timeout is scheduled.
     @usableFromInline
@@ -51,7 +51,7 @@ extension ConnectionPool {
     internal init(
       deadline: NIODeadline,
       promise: EventLoopPromise<Channel>,
-      channelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture<Void>
+      channelInitializer: @escaping (Channel) -> EventLoopFuture<Void>
     ) {
       self._deadline = deadline
       self._promise = promise
@@ -83,7 +83,7 @@ extension ConnectionPool {
 
     /// Succeed the waiter with the given multiplexer.
     @usableFromInline
-    internal func succeed(with multiplexer: NIOHTTP2Handler.StreamMultiplexer) {
+    internal func succeed(with multiplexer: HTTP2StreamMultiplexer) {
       self._scheduledTimeout?.cancel()
       self._scheduledTimeout = nil
       multiplexer.createStreamChannel(promise: self._promise, self._channelInitializer)