Add an http/2 connection delegate (#1164)

Motivation:

As part of #1034 we need some way to notify a connection pool about the
number of active streams on a connection and when we receive a change to
SETTINGS_MAX_CONCURRENT_STREAMS.

Modifications:

Add an internal `HTTP2ConnectionDelegate` protocol and allow it to be
plugged in to the connectivity state monitor. Wire it up to the idle
handler where we recieve notifications of the relevant events.

Result:

The connection pool can be notified when an http/2 stream is closed or
when SETTINGS_MAX_CONCURRENT_STREAMS changes.

Sources/GRPC/ConnectivityState.swift

@@ -74,6 +74,7 @@ public class ConnectivityStateMonitor {
 
   private let delegateLock = Lock()
   private var _delegate: ConnectivityStateDelegate?
+  private var _http2Delegate: HTTP2ConnectionDelegate?
   private let delegateCallbackQueue: DispatchQueue
 
   /// Creates a new connectivity state monitor.
@@ -82,7 +83,7 @@ public class ConnectivityStateMonitor {
   /// - Parameter queue: The `DispatchQueue` on which the delegate will be called.
   init(delegate: ConnectivityStateDelegate?, queue: DispatchQueue?) {
     self._delegate = delegate
-    self.delegateCallbackQueue = queue ?? DispatchQueue(label: "io.grpc.connectivity")
+    self.delegateCallbackQueue = DispatchQueue(label: "io.grpc.connectivity", target: queue)
   }
 
   /// The current state of connectivity.
@@ -105,7 +106,9 @@ public class ConnectivityStateMonitor {
       }
     }
   }
+}
 
+extension ConnectivityStateMonitor {
   internal func updateState(to newValue: ConnectivityState, logger: Logger) {
     let change: (ConnectivityState, ConnectivityState)? = self.stateLock.withLock {
       let oldValue = self._state
@@ -140,3 +143,34 @@ public class ConnectivityStateMonitor {
     }
   }
 }
+
+extension ConnectivityStateMonitor: HTTP2ConnectionDelegate {
+  internal final var http2Delegate: HTTP2ConnectionDelegate? {
+    get {
+      return self.delegateLock.withLock {
+        return self._http2Delegate
+      }
+    }
+    set {
+      self.delegateLock.withLockVoid {
+        self._http2Delegate = newValue
+      }
+    }
+  }
+
+  internal final func streamClosed() {
+    if let delegate = self.http2Delegate {
+      self.delegateCallbackQueue.async {
+        delegate.streamClosed()
+      }
+    }
+  }
+
+  internal final func maxConcurrentStreamsChanged(_ maxConcurrentStreams: Int) {
+    if let delegate = self.http2Delegate {
+      self.delegateCallbackQueue.async {
+        delegate.maxConcurrentStreamsChanged(maxConcurrentStreams)
+      }
+    }
+  }
+}

Sources/GRPC/GRPCIdleHandler.swift

@@ -130,6 +130,12 @@ internal final class GRPCIdleHandler: ChannelInboundHandler {
       }
     }
 
+    // Max concurrent streams changed.
+    if let manager = self.mode.connectionManager,
+      let maxConcurrentStreams = operations.maxConcurrentStreamsChange {
+      manager.monitor.maxConcurrentStreamsChanged(maxConcurrentStreams)
+    }
+
     // Handle idle timeout creation/cancellation.
     if let idleTask = operations.idleTask {
       switch idleTask {
@@ -222,6 +228,9 @@ internal final class GRPCIdleHandler: ChannelInboundHandler {
     } else if let closed = event as? StreamClosedEvent {
       self.perform(operations: self.stateMachine.streamClosed(withID: closed.streamID))
       self.handlePingAction(self.pingHandler.streamClosed())
+      if let manager = self.mode.connectionManager {
+        manager.monitor.streamClosed()
+      }
       context.fireUserInboundEventTriggered(event)
     } else if event is ChannelShouldQuiesceEvent {
       self.perform(operations: self.stateMachine.initiateGracefulShutdown())

Sources/GRPC/GRPCIdleHandlerStateMachine.swift

@@ -186,6 +186,9 @@ struct GRPCIdleHandlerStateMachine {
     /// Whether the channel should be closed.
     private(set) var shouldCloseChannel: Bool
 
+    /// The new value of 'SETTINGS_MAX_CONCURRENT_STREAMS'.
+    private(set) var maxConcurrentStreamsChange: Int?
+
     fileprivate static let none = Operations()
 
     fileprivate mutating func sendGoAwayFrame(lastPeerInitiatedStreamID streamID: HTTP2StreamID) {
@@ -208,6 +211,10 @@ struct GRPCIdleHandlerStateMachine {
       self.connectionManagerEvent = event
     }
 
+    fileprivate mutating func maxConcurrentStreamsChanged(_ count: Int) {
+      self.maxConcurrentStreamsChange = count
+    }
+
     private init() {
       self.connectionManagerEvent = nil
       self.idleTask = nil
@@ -502,6 +509,7 @@ struct GRPCIdleHandlerStateMachine {
 
       // Update max concurrent streams.
       if let maxStreams = settings.last(where: { $0.parameter == .maxConcurrentStreams })?.value {
+        operations.maxConcurrentStreamsChanged(maxStreams)
         state.maxConcurrentStreams = maxStreams
       }
       self.state = .operating(state)
@@ -509,6 +517,7 @@ struct GRPCIdleHandlerStateMachine {
     case var .waitingToIdle(state):
       // Update max concurrent streams.
       if let maxStreams = settings.last(where: { $0.parameter == .maxConcurrentStreams })?.value {
+        operations.maxConcurrentStreamsChanged(maxStreams)
         state.maxConcurrentStreams = maxStreams
       }
       self.state = .waitingToIdle(state)

Sources/GRPC/HTTP2ConnectionDelegate.swift

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2021, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import NIO
+import NIOHTTP2
+
+internal protocol HTTP2ConnectionDelegate {
+  /// An HTTP/2 stream was closed.
+  func streamClosed()
+
+  /// The value of 'SETTINGS_MAX_CONCURRENT_STREAMS' changed.
+  /// - Parameter maxConcurrentStreams: The new value for 'SETTINGS_MAX_CONCURRENT_STREAMS'.
+  func maxConcurrentStreamsChanged(_ maxConcurrentStreams: Int)
+}

Tests/GRPCTests/ConnectivityStateMonitorTests.swift

@@ -43,4 +43,35 @@ class ConnectivityStateMonitorTests: GRPCTestCase {
 
     recorder.waitForExpectedChanges(timeout: .seconds(1))
   }
+
+  func testHTTP2Delegate() {
+    let http2Delegate = RecordingHTTP2Delegate()
+    let queue = DispatchQueue(label: "io.grpc.testing")
+    let monitor = ConnectivityStateMonitor(delegate: nil, queue: queue)
+    monitor.http2Delegate = http2Delegate
+
+    monitor.streamClosed()
+    monitor.streamClosed()
+    monitor.streamClosed()
+
+    monitor.maxConcurrentStreamsChanged(31)
+    monitor.maxConcurrentStreamsChanged(41)
+    monitor.maxConcurrentStreamsChanged(49)
+
+    XCTAssertEqual(queue.sync { http2Delegate.streamsClosed }, 3)
+    XCTAssertEqual(queue.sync { http2Delegate.maxConcurrentStreamsChanges }, [31, 41, 49])
+  }
+}
+
+internal final class RecordingHTTP2Delegate: HTTP2ConnectionDelegate {
+  private(set) var streamsClosed = 0
+  private(set) var maxConcurrentStreamsChanges: [Int] = []
+
+  internal func streamClosed() {
+    self.streamsClosed += 1
+  }
+
+  internal func maxConcurrentStreamsChanged(_ maxConcurrentStreams: Int) {
+    self.maxConcurrentStreamsChanges.append(maxConcurrentStreams)
+  }
 }

Tests/GRPCTests/HTTP2DelegateTests.swift

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2021, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import EchoImplementation
+import EchoModel
+@testable import GRPC
+import NIO
+import XCTest
+
+final class HTTP2ConnectionDelegateTests: GRPCTestCase {
+  private var group: EventLoopGroup!
+  private var server: Server!
+  private var connection: ClientConnection!
+  private let queue = DispatchQueue(label: "io.grpc.testing")
+
+  override func setUp() {
+    super.setUp()
+    self.group = MultiThreadedEventLoopGroup(numberOfThreads: 2)
+    self.server = try! Server.insecure(group: self.group)
+      .withLogger(self.serverLogger)
+      .withServiceProviders([EchoProvider()])
+      .bind(host: "127.0.0.1", port: 0)
+      .wait()
+
+    self.connection = ClientConnection.insecure(group: self.group)
+      .withBackgroundActivityLogger(self.clientLogger)
+      // The http/2 delegate is internal but uses the same queue as the connectivity state delegate,
+      // so this looks odd but is fine.
+      .withConnectivityStateDelegate(nil, executingOn: self.queue)
+      .connect(host: "127.0.0.1", port: self.server!.channel.localAddress!.port!)
+  }
+
+  override func tearDown() {
+    XCTAssertNoThrow(try self.connection.close().wait())
+    XCTAssertNoThrow(try self.server.close().wait())
+    XCTAssertNoThrow(try self.group.syncShutdownGracefully())
+    super.tearDown()
+  }
+
+  func testDelegate() {
+    let http2Delegate = RecordingHTTP2Delegate()
+    self.connection.connectivity.http2Delegate = http2Delegate
+
+    let echo = Echo_EchoClient(channel: self.connection)
+
+    // Fire off a handful of RPCs.
+    for _ in 0 ..< 10 {
+      let get = echo.get(.with { $0.text = "" })
+      XCTAssertNoThrow(try get.status.wait())
+    }
+
+    // 10 RPCs, 10 streams closed.
+    XCTAssertEqual(self.queue.sync { http2Delegate.streamsClosed }, 10)
+    XCTAssertEqual(self.queue.sync { http2Delegate.maxConcurrentStreamsChanges }, [100])
+  }
+}