Add a connection pool delegate (#1515)

Motivation:

It can be useful to observe what the underlying connection pool is doing
over time. This could, for example, be instrumenting it to understand
how many streams are active at a given time or to understand when a
connection is in the processing of being brought up.

Modifications:

Add new API:
- `GRPCConnectionPoolDelegate` which users can implement and configure
  on the `GRPCChannelPool` to receive notifications about the connection
  pool state.
- `GRPCConnectionID`, an opaque ID to distinguish between different
  connections in the pool.

Modify the connection pool:
- Per-EventLoop pools now hold on to quiescing connections and track
  additional state, including whether a connection is quiescing and how
  many streams are currently open on a connection. By contrast the pool
  manager (which manages a number of these per-EventLoop pools) tracks
  reserved streams (which are not necessarily open). The delegate tracks
  _opened_ streams rather than _reserved_ streams (as reserved streams
  are not allocated to a connection but to an any connection running on
  an appropriate event-loop).
- Wire through the new delegate and call out to it in appropriate
  places.
- Add a stream opened function to the internal H2 delegate
- Expose various pieces of state from the connection manager

Result:

Users can instrument the underlying connection pool.

Author: glbrntt

Sources/GRPC/ConnectionManager+Delegates.swift

@@ -35,6 +35,12 @@ internal protocol ConnectionManagerConnectivityDelegate {
 }
 
 internal protocol ConnectionManagerHTTP2Delegate {
+  /// An HTTP/2 stream was opened.
+  ///
+  /// - Parameters:
+  ///   - connectionManager: The connection manager reporting the opened stream.
+  func streamOpened(_ connectionManager: ConnectionManager)
+
   /// An HTTP/2 stream was closed.
   ///
   /// - Parameters:

Sources/GRPC/ConnectionManager.swift

@@ -242,6 +242,17 @@ internal final class ConnectionManager {
     }
   }
 
+  /// Returns whether the state is 'shutdown'.
+  private var isShutdown: Bool {
+    self.eventLoop.assertInEventLoop()
+    switch self.state {
+    case .shutdown:
+      return true
+    case .idle, .connecting, .transientFailure, .active, .ready:
+      return false
+    }
+  }
+
   /// Returns the `HTTP2StreamMultiplexer` from the 'ready' state or `nil` if it is not available.
   private var multiplexer: HTTP2StreamMultiplexer? {
     self.eventLoop.assertInEventLoop()
@@ -582,6 +593,31 @@ internal final class ConnectionManager {
     }
   }
 
+  /// Registers a callback which fires when the current active connection is closed.
+  ///
+  /// If there is a connection, the callback will be invoked with `true` when the connection is
+  /// closed. Otherwise the callback is invoked with `false`.
+  internal func onCurrentConnectionClose(_ onClose: @escaping (Bool) -> Void) {
+    if self.eventLoop.inEventLoop {
+      self._onCurrentConnectionClose(onClose)
+    } else {
+      self.eventLoop.execute {
+        self._onCurrentConnectionClose(onClose)
+      }
+    }
+  }
+
+  private func _onCurrentConnectionClose(_ onClose: @escaping (Bool) -> Void) {
+    self.eventLoop.assertInEventLoop()
+
+    switch self.state {
+    case let .ready(state):
+      state.channel.closeFuture.whenComplete { _ in onClose(true) }
+    case .idle, .connecting, .active, .transientFailure, .shutdown:
+      onClose(false)
+    }
+  }
+
   // MARK: - State changes from the channel handler.
 
   /// The channel caught an error. Hold on to it until the channel becomes inactive, it may provide
@@ -807,6 +843,11 @@ internal final class ConnectionManager {
     }
   }
 
+  internal func streamOpened() {
+    self.eventLoop.assertInEventLoop()
+    self.http2Delegate?.streamOpened(self)
+  }
+
   internal func streamClosed() {
     self.eventLoop.assertInEventLoop()
     self.http2Delegate?.streamClosed(self)
@@ -1001,6 +1042,11 @@ extension ConnectionManager {
       return self.manager.isIdle
     }
 
+    /// Returne `true` if the connection is in the shutdown state.
+    internal var isShutdown: Bool {
+      return self.manager.isShutdown
+    }
+
     /// Returns the `multiplexer` from a connection in the `ready` state or `nil` if it is any
     /// other state.
     internal var multiplexer: HTTP2StreamMultiplexer? {

Sources/GRPC/ConnectionPool/ConnectionManagerID.swift

@@ -15,7 +15,7 @@
  */
 
 @usableFromInline
-internal struct ConnectionManagerID: Hashable, CustomStringConvertible {
+internal struct ConnectionManagerID: Hashable, CustomStringConvertible, GRPCSendable {
   @usableFromInline
   internal let _id: ObjectIdentifier
 

Sources/GRPC/ConnectionPool/ConnectionPool+PerConnectionState.swift

@@ -26,8 +26,32 @@ extension ConnectionPool {
     @usableFromInline
     internal var _availability: StreamAvailability?
 
+    @usableFromInline
+    internal var isQuiescing: Bool {
+      get {
+        return self._availability?.isQuiescing ?? false
+      }
+      set {
+        self._availability?.isQuiescing = true
+      }
+    }
+
     @usableFromInline
     internal struct StreamAvailability {
+      @usableFromInline
+      struct Utilization {
+        @usableFromInline
+        var used: Int
+        @usableFromInline
+        var capacity: Int
+
+        @usableFromInline
+        init(used: Int, capacity: Int) {
+          self.used = used
+          self.capacity = capacity
+        }
+      }
+
       @usableFromInline
       var multiplexer: HTTP2StreamMultiplexer
       /// Maximum number of available streams.
@@ -36,24 +60,39 @@ extension ConnectionPool {
       /// Number of streams reserved.
       @usableFromInline
       var reserved: Int = 0
+      /// Number of streams opened.
+      @usableFromInline
+      var open: Int = 0
+      @usableFromInline
+      var isQuiescing = false
       /// Number of available streams.
       @usableFromInline
       var available: Int {
-        return self.maxAvailable - self.reserved
+        return self.isQuiescing ? 0 : self.maxAvailable - self.reserved
       }
 
       /// Increment the reserved streams and return the multiplexer.
       @usableFromInline
       mutating func reserve() -> HTTP2StreamMultiplexer {
+        assert(!self.isQuiescing)
         self.reserved += 1
         return self.multiplexer
       }
 
+      @usableFromInline
+      mutating func opened() -> Utilization {
+        self.open += 1
+        return .init(used: self.open, capacity: self.maxAvailable)
+      }
+
       /// Decrement the reserved streams by one.
       @usableFromInline
-      mutating func `return`() {
+      mutating func `return`() -> Utilization {
         self.reserved -= 1
+        self.open -= 1
         assert(self.reserved >= 0)
+        assert(self.open >= 0)
+        return .init(used: self.open, capacity: self.maxAvailable)
       }
     }
 
@@ -92,10 +131,15 @@ extension ConnectionPool {
       return self._availability?.reserve()
     }
 
+    @usableFromInline
+    internal mutating func openedStream() -> PerConnectionState.StreamAvailability.Utilization? {
+      return self._availability?.opened()
+    }
+
     /// Return a reserved stream to the connection.
     @usableFromInline
-    internal mutating func returnStream() {
-      self._availability?.return()
+    internal mutating func returnStream() -> PerConnectionState.StreamAvailability.Utilization? {
+      return self._availability?.return()
     }
 
     /// Update the maximum concurrent streams available on the connection, marking it as available