Add async implementation of QPSBenchmark (#1471)

Motivation:

We want to provide an async/await version of the QPSBenchmark implementation, and be able to choose whether we want to use that or the ELF version when running the performance tests via the gRPC JSON driver.

Modifications:

- Added full async/await-based implementations of the worker, server, and client.
- Created a new command line flag (`--use-async`) to choose between the EventLoopFuture and async/await versions when running the tests.

Result:

QPSBenchmark can optionally be executed using an async/await implementation.

Author: gjcairo

Performance/QPSBenchmark/Package.swift

@@ -25,16 +25,15 @@ let package = Package(
   ],
   dependencies: [
     .package(path: "../../"),
-    .package(url: "https://github.com/apple/swift-nio.git", from: "2.32.0"),
-    .package(url: "https://github.com/apple/swift-log.git", from: "1.4.0"),
-    .package(url: "https://github.com/apple/swift-argument-parser.git", from: "1.0.0"),
+    .package(url: "https://github.com/apple/swift-nio.git", from: "2.41.0"),
+    .package(url: "https://github.com/apple/swift-log.git", from: "1.4.3"),
+    .package(url: "https://github.com/apple/swift-argument-parser.git", from: "1.1.1"),
     .package(url: "https://github.com/apple/swift-atomics.git", from: "1.0.2"),
     .package(
       url: "https://github.com/swift-server/swift-service-lifecycle.git",
       from: "1.0.0-alpha"
     ),
     .package(
-      name: "SwiftProtobuf",
       url: "https://github.com/apple/swift-protobuf.git",
       from: "1.19.0"
     ),
@@ -51,7 +50,7 @@ let package = Package(
         .product(name: "ArgumentParser", package: "swift-argument-parser"),
         .product(name: "Logging", package: "swift-log"),
         .product(name: "Lifecycle", package: "swift-service-lifecycle"),
-        .product(name: "SwiftProtobuf", package: "SwiftProtobuf"),
+        .product(name: "SwiftProtobuf", package: "swift-protobuf"),
         .target(name: "BenchmarkUtils"),
       ],
       exclude: [

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/Async/AsyncClientProtocol.swift

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2022, 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 GRPC
+import NIOCore
+
+/// Protocol which async clients must implement.
+protocol AsyncQPSClient {
+  /// Start the execution of the client.
+  func startClient()
+
+  /// Send the status of the current test
+  /// - parameters:
+  ///     - reset: Indicates if the stats collection should be reset after publication or not.
+  ///     - responseStream: the response stream to write the response to.
+  func sendStatus(
+    reset: Bool,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ClientStatus>
+  ) async throws
+
+  /// Shut down the client.
+  func shutdown() async throws
+}

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/Async/AsyncPingPongRequestMaker.swift

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2022, 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 Atomics
+import Foundation
+import GRPC
+import Logging
+import NIOCore
+
+/// Makes streaming requests and listens to responses ping-pong style.
+/// Iterations can be limited by config.
+/// Class is marked as `@unchecked Sendable` because `ManagedAtomic<Bool>` doesn't conform
+/// to `Sendable`, but we know it's safe.
+final class AsyncPingPongRequestMaker: AsyncRequestMaker, @unchecked Sendable {
+  private let client: Grpc_Testing_BenchmarkServiceAsyncClient
+  private let requestMessage: Grpc_Testing_SimpleRequest
+  private let logger: Logger
+  private let stats: StatsWithLock
+
+  /// If greater than zero gives a limit to how many messages are exchanged before termination.
+  private let messagesPerStream: Int
+  /// Stops more requests being made after stop is requested.
+  private let stopRequested = ManagedAtomic<Bool>(false)
+
+  /// Initialiser to gather requirements.
+  /// - Parameters:
+  ///    - config: config from the driver describing what to do.
+  ///    - client: client interface to the server.
+  ///    - requestMessage: Pre-made request message to use possibly repeatedly.
+  ///    - logger: Where to log useful diagnostics.
+  ///    - stats: Where to record statistics on latency.
+  init(
+    config: Grpc_Testing_ClientConfig,
+    client: Grpc_Testing_BenchmarkServiceAsyncClient,
+    requestMessage: Grpc_Testing_SimpleRequest,
+    logger: Logger,
+    stats: StatsWithLock
+  ) {
+    self.client = client
+    self.requestMessage = requestMessage
+    self.logger = logger
+    self.stats = stats
+
+    self.messagesPerStream = Int(config.messagesPerStream)
+  }
+
+  /// Initiate a request sequence to the server - in this case the sequence is streaming requests to the server and waiting
+  /// to see responses before repeating ping-pong style.  The number of iterations can be limited by config.
+  func makeRequest() async throws {
+    var startTime = grpcTimeNow()
+    var messagesSent = 0
+
+    let streamingCall = self.client.makeStreamingCallCall()
+    var responseStream = streamingCall.responseStream.makeAsyncIterator()
+    while !self.stopRequested.load(ordering: .relaxed),
+          self.messagesPerStream == 0 || messagesSent < self.messagesPerStream {
+      try await streamingCall.requestStream.send(self.requestMessage)
+      let _ = try await responseStream.next()
+      let endTime = grpcTimeNow()
+      self.stats.add(latency: endTime - startTime)
+      messagesSent += 1
+      startTime = endTime
+    }
+  }
+
+  /// Request termination of the request-response sequence.
+  func requestStop() {
+    self.logger.info("AsyncPingPongRequestMaker stop requested")
+    // Flag stop as requested - this will prevent any more requests being made.
+    self.stopRequested.store(true, ordering: .relaxed)
+  }
+}