Drop user context when closing client interceptor pipeline (#1279)

Motivation:

Interceptor pipelines hold arrays of interceptor contexts. Each context
maintains a reference back to the pipeline. However the client
interceptor pipeline never broke this reference cycle when closing the
pipeline which means that using client interceptors introduces a memory
leak.

Modifications:

- Explicitly remove all user contexts when the client interceptor pipeline is
  torn down (the server pipeline already does this so is unaffected)
- Add allocation tests so we can see the overhead of using interceptors
  (and, more importantly, whether allocations are freed)

Result:

Using client interceptors no longer leaks.

.github/workflows/ci.yaml

@@ -57,6 +57,8 @@ jobs:
  MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_1_small_request: 67000
  MAX_ALLOCS_ALLOWED_embedded_server_unary_1k_rpcs_1_small_request: 63000
  MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 204000
+ MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 211000
+ MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 211000
  - image: swift:5.3-focal
  env:
  MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 504000
@@ -65,6 +67,8 @@ jobs:
  MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_1_small_request: 67000
  MAX_ALLOCS_ALLOWED_embedded_server_unary_1k_rpcs_1_small_request: 63000
  MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 205000
+ MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 212000
+ MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 212000
  - image: swift:5.2-bionic
  env:
  MAX_ALLOCS_ALLOWED_bidi_1k_rpcs_10_requests: 515000
@@ -73,6 +77,8 @@ jobs:
  MAX_ALLOCS_ALLOWED_embedded_server_bidi_1k_rpcs_1_small_request: 67000
  MAX_ALLOCS_ALLOWED_embedded_server_unary_1k_rpcs_1_small_request: 63000
  MAX_ALLOCS_ALLOWED_unary_1k_ping_pong: 206000
+ MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_client: 213000
+ MAX_ALLOCS_ALLOWED_unary_1k_ping_pong_interceptors_server: 213000
  name: Performance Tests on ${{ matrix.image }}
  runs-on: ubuntu-latest
  container:

Performance/allocations/tests/shared/Common.swift

@@ -19,10 +19,11 @@ import NIO
 func makeEchoServer(
  group: EventLoopGroup,
  host: String = "127.0.0.1",
- port: Int = 0
+ port: Int = 0,
+ interceptors: Echo_EchoServerInterceptorFactoryProtocol? = nil
 ) -> EventLoopFuture<Server> {
  return Server.insecure(group: group)
- .withServiceProviders([MinimalEchoProvider()])
+ .withServiceProviders([MinimalEchoProvider(interceptors: interceptors)])
  .bind(host: host, port: port)
 }
 
@@ -34,3 +35,88 @@ func makeClientConnection(
  return ClientConnection.insecure(group: group)
  .connect(host: host, port: port)
 }
+
+func makeEchoClientInterceptors(count: Int) -> Echo_EchoClientInterceptorFactoryProtocol? {
+ let factory = EchoClientInterceptors()
+ for _ in 0 ..< count {
+ factory.register { NoOpEchoClientInterceptor() }
+ }
+ return factory
+}
+
+func makeEchoServerInterceptors(count: Int) -> Echo_EchoServerInterceptorFactoryProtocol? {
+ let factory = EchoServerInterceptors()
+ for _ in 0 ..< count {
+ factory.register { NoOpEchoServerInterceptor() }
+ }
+ return factory
+}
+
+final class EchoClientInterceptors: Echo_EchoClientInterceptorFactoryProtocol {
+ internal typealias Factory = () -> ClientInterceptor<Echo_EchoRequest, Echo_EchoResponse>
+ private var factories: [Factory] = []
+
+ internal init(_ factories: Factory...) {
+ self.factories = factories
+ }
+
+ internal func register(_ factory: @escaping Factory) {
+ self.factories.append(factory)
+ }
+
+ private func makeInterceptors() -> [ClientInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.factories.map { $0() }
+ }
+
+ func makeGetInterceptors() -> [ClientInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.makeInterceptors()
+ }
+
+ func makeExpandInterceptors() -> [ClientInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.makeInterceptors()
+ }
+
+ func makeCollectInterceptors() -> [ClientInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.makeInterceptors()
+ }
+
+ func makeUpdateInterceptors() -> [ClientInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.makeInterceptors()
+ }
+}
+
+internal final class EchoServerInterceptors: Echo_EchoServerInterceptorFactoryProtocol {
+ internal typealias Factory = () -> ServerInterceptor<Echo_EchoRequest, Echo_EchoResponse>
+ private var factories: [Factory] = []
+
+ internal init(_ factories: Factory...) {
+ self.factories = factories
+ }
+
+ internal func register(_ factory: @escaping Factory) {
+ self.factories.append(factory)
+ }
+
+ private func makeInterceptors() -> [ServerInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.factories.map { $0() }
+ }
+
+ func makeGetInterceptors() -> [ServerInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.makeInterceptors()
+ }
+
+ func makeExpandInterceptors() -> [ServerInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.makeInterceptors()
+ }
+
+ func makeCollectInterceptors() -> [ServerInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.makeInterceptors()
+ }
+
+ func makeUpdateInterceptors() -> [ServerInterceptor<Echo_EchoRequest, Echo_EchoResponse>] {
+ return self.makeInterceptors()
+ }
+}
+
+final class NoOpEchoClientInterceptor: ClientInterceptor<Echo_EchoRequest, Echo_EchoResponse> {}
+final class NoOpEchoServerInterceptor: ServerInterceptor<Echo_EchoRequest, Echo_EchoResponse> {}

Performance/allocations/tests/test_unary_1k_ping_pong.swift

@@ -23,15 +23,31 @@ class UnaryPingPongBenchmark: Benchmark {
  private var group: EventLoopGroup!
  private var server: Server!
  private var client: ClientConnection!
+ private let clientInterceptors: Echo_EchoClientInterceptorFactoryProtocol?
+ private let serverInterceptors: Echo_EchoServerInterceptorFactoryProtocol?
 
- init(rpcs: Int, request: String) {
+ init(
+ rpcs: Int,
+ request: String,
+ clientInterceptors: Int = 0,
+ serverInterceptors: Int = 0
+ ) {
  self.rpcs = rpcs
  self.request = .with { $0.text = request }
+ self.clientInterceptors = clientInterceptors > 0
+ ? makeEchoClientInterceptors(count: clientInterceptors)
+ : nil
+ self.serverInterceptors = serverInterceptors > 0
+ ? makeEchoServerInterceptors(count: serverInterceptors)
+ : nil
  }
 
  func setUp() throws {
  self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
- self.server = try makeEchoServer(group: self.group).wait()
+ self.server = try makeEchoServer(
+ group: self.group,
+ interceptors: self.serverInterceptors
+ ).wait()
  self.client = makeClientConnection(
  group: self.group,
  port: self.server.channel.localAddress!.port!
@@ -45,7 +61,7 @@ class UnaryPingPongBenchmark: Benchmark {
  }
 
  func run() throws -> Int {
- let echo = Echo_EchoClient(channel: self.client)
+ let echo = Echo_EchoClient(channel: self.client, interceptors: self.clientInterceptors)
  var responseLength = 0
 
  for _ in 0 ..< self.rpcs {
@@ -63,4 +79,14 @@ func run(identifier: String) {
  let benchmark = UnaryPingPongBenchmark(rpcs: 1000, request: "")
  return try! benchmark.runOnce()
  }
+
+ measure(identifier: identifier + "_interceptors_server") {
+ let benchmark = UnaryPingPongBenchmark(rpcs: 1000, request: "", serverInterceptors: 5)
+ return try! benchmark.runOnce()
+ }
+
+ measure(identifier: identifier + "_interceptors_client") {
+ let benchmark = UnaryPingPongBenchmark(rpcs: 1000, request: "", clientInterceptors: 5)
+ return try! benchmark.runOnce()
+ }
 }

Sources/GRPC/Interceptor/ClientInterceptorPipeline.swift

@@ -443,6 +443,9 @@ extension ClientInterceptorPipeline {
  self._scheduledClose?.cancel()
  self._scheduledClose = nil
 
+ // Drop the contexts since they reference us.
+ self._userContexts.removeAll()
+
  // Cancel the transport.
  self._onCancel?(nil)
 

Sources/GRPCPerformanceTests/Benchmarks/MinimalEchoProvider.swift

@@ -19,7 +19,11 @@ import NIO
 /// The echo provider that comes with the example does some string processing, we'll avoid some of
 /// that here so we're looking at the right things.
 public class MinimalEchoProvider: Echo_EchoProvider {
- public let interceptors: Echo_EchoServerInterceptorFactoryProtocol? = nil
+ public let interceptors: Echo_EchoServerInterceptorFactoryProtocol?
+
+ public init(interceptors: Echo_EchoServerInterceptorFactoryProtocol? = nil) {
+ self.interceptors = interceptors
+ }
 
  public func get(
  request: Echo_EchoRequest,