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.

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)
+  }
+}

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

@@ -0,0 +1,293 @@
+/*
+ * 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 BenchmarkUtils
+import Foundation
+import GRPC
+import Logging
+import NIOConcurrencyHelpers
+import NIOCore
+import NIOPosix
+
+/// Client to make a series of asynchronous calls.
+final class AsyncQPSClientImpl<RequestMakerType: AsyncRequestMaker>: AsyncQPSClient {
+  private let logger = Logger(label: "AsyncQPSClientImpl")
+
+  private let eventLoopGroup: MultiThreadedEventLoopGroup
+  private let threadCount: Int
+  private let channelRepeaters: [ChannelRepeater]
+
+  private var statsPeriodStart: DispatchTime
+  private var cpuStatsPeriodStart: CPUTime
+
+  /// Initialise a client to send requests.
+  /// - parameters:
+  ///      - config: Config from the driver specifying how the client should behave.
+  init(config: Grpc_Testing_ClientConfig) throws {
+    // Setup threads
+    let threadCount = config.threadsToUse()
+    self.threadCount = threadCount
+    self.logger.info("Sizing AsyncQPSClientImpl", metadata: ["threads": "\(threadCount)"])
+    let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: threadCount)
+    self.eventLoopGroup = eventLoopGroup
+
+    // Parse possible invalid targets before code with side effects.
+    let serverTargets = try config.parsedServerTargets()
+    precondition(serverTargets.count > 0)
+
+    // Start recording stats.
+    self.statsPeriodStart = grpcTimeNow()
+    self.cpuStatsPeriodStart = getResourceUsage()
+
+    let requestMessage = try AsyncQPSClientImpl
+      .makeClientRequest(payloadConfig: config.payloadConfig)
+
+    // Start the requested number of channels.
+    self.channelRepeaters = (0 ..< Int(config.clientChannels)).map { channelNumber in
+      ChannelRepeater(
+        target: serverTargets[channelNumber % serverTargets.count],
+        requestMessage: requestMessage,
+        config: config,
+        eventLoop: eventLoopGroup.next()
+      )
+    }
+  }
+
+  /// Start the execution of the client.
+  func startClient() {
+    Task {
+      try await withThrowingTaskGroup(of: Void.self) { group in
+        for repeater in self.channelRepeaters {
+          group.addTask {
+            try await repeater.start()
+          }
+        }
+        try await group.waitForAll()
+      }
+    }
+  }
+
+  /// Send current status back to the driver process.
+  /// - parameters:
+  ///     - reset: Should the stats reset after being sent.
+  ///     - context: Calling context to allow results to be sent back to the driver.
+  func sendStatus(
+    reset: Bool,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ClientStatus>
+  ) async throws {
+    let currentTime = grpcTimeNow()
+    let currentResourceUsage = getResourceUsage()
+    var result = Grpc_Testing_ClientStatus()
+    result.stats.timeElapsed = (currentTime - self.statsPeriodStart).asSeconds()
+    result.stats.timeSystem = currentResourceUsage.systemTime - self.cpuStatsPeriodStart
+      .systemTime
+    result.stats.timeUser = currentResourceUsage.userTime - self.cpuStatsPeriodStart.userTime
+    result.stats.cqPollCount = 0
+
+    // Collect stats from each of the channels.
+    var latencyHistogram = Histogram()
+    var statusCounts = StatusCounts()
+    for channelRepeater in self.channelRepeaters {
+      let stats = channelRepeater.getStats(reset: reset)
+      try! latencyHistogram.merge(source: stats.latencies)
+      statusCounts.merge(source: stats.statuses)
+    }
+    result.stats.latencies = Grpc_Testing_HistogramData(from: latencyHistogram)
+    result.stats.requestResults = statusCounts.toRequestResultCounts()
+    self.logger.info("Sending client status")
+    try await responseStream.send(result)
+
+    if reset {
+      self.statsPeriodStart = currentTime
+      self.cpuStatsPeriodStart = currentResourceUsage
+    }
+  }
+
+  /// Shut down the service.
+  func shutdown() async throws {
+    await withThrowingTaskGroup(of: Void.self) { group in
+      for repeater in self.channelRepeaters {
+        group.addTask {
+          do {
+            try await repeater.stop()
+          } catch {
+            self.logger.warning(
+              "A channel repeater could not be stopped",
+              metadata: ["error": "\(error)"]
+            )
+          }
+        }
+      }
+    }
+  }
+
+  /// Make a request which can be sent to the server.
+  private static func makeClientRequest(
+    payloadConfig: Grpc_Testing_PayloadConfig
+  ) throws -> Grpc_Testing_SimpleRequest {
+    if let payload = payloadConfig.payload {
+      switch payload {
+      case .bytebufParams:
+        throw GRPCStatus(code: .invalidArgument, message: "Byte buffer not supported.")
+      case let .simpleParams(simpleParams):
+        var result = Grpc_Testing_SimpleRequest()
+        result.responseType = .compressable
+        result.responseSize = simpleParams.respSize
+        result.payload.type = .compressable
+        let size = Int(simpleParams.reqSize)
+        let body = Data(count: size)
+        result.payload.body = body
+        return result
+      case .complexParams:
+        throw GRPCStatus(
+          code: .invalidArgument,
+          message: "Complex params not supported."
+        )
+      }
+    } else {
+      // Default - simple proto without payloads.
+      var result = Grpc_Testing_SimpleRequest()
+      result.responseType = .compressable
+      result.responseSize = 0
+      result.payload.type = .compressable
+      return result
+    }
+  }
+
+  /// Class to manage a channel.  Repeatedly makes requests on that channel and records what happens.
+  /// /// Class is marked as `@unchecked Sendable` because `ManagedAtomic<Bool>` doesn't conform
+  /// to `Sendable`, but we know it's safe.
+  private final class ChannelRepeater: @unchecked Sendable {
+    private let channel: GRPCChannel
+    private let eventLoop: EventLoop
+    private let maxPermittedOutstandingRequests: Int
+
+    private let stats: StatsWithLock
+
+    /// Succeeds after a stop has been requested and all outstanding requests have completed.
+    private let stopComplete: EventLoopPromise<Void>
+
+    private let running = ManagedAtomic<Bool>(false)
+
+    private let requestMaker: RequestMakerType
+
+    init(
+      target: ConnectionTarget,
+      requestMessage: Grpc_Testing_SimpleRequest,
+      config: Grpc_Testing_ClientConfig,
+      eventLoop: EventLoop
+    ) {
+      self.eventLoop = eventLoop
+      // 'try!' is fine; it'll only throw if we can't make an SSL context
+      // TODO: Support TLS if requested.
+      self.channel = try! GRPCChannelPool.with(
+        target: target,
+        transportSecurity: .plaintext,
+        eventLoopGroup: eventLoop
+      )
+
+      let logger = Logger(label: "ChannelRepeater")
+      let client = Grpc_Testing_BenchmarkServiceAsyncClient(channel: self.channel)
+      self.maxPermittedOutstandingRequests = Int(config.outstandingRpcsPerChannel)
+      self.stopComplete = eventLoop.makePromise()
+      self.stats = StatsWithLock()
+
+      self.requestMaker = RequestMakerType(
+        config: config,
+        client: client,
+        requestMessage: requestMessage,
+        logger: logger,
+        stats: self.stats
+      )
+    }
+
+    /// Launch as many requests as allowed on the channel. Must only be called once.
+    private func launchRequests() async throws {
+      let exchangedRunning = self.running.compareExchange(
+        expected: false,
+        desired: true,
+        ordering: .relaxed
+      )
+      precondition(exchangedRunning.exchanged, "launchRequests should only be called once")
+
+      try await withThrowingTaskGroup(of: Void.self) { group in
+        for _ in 0 ..< self.maxPermittedOutstandingRequests {
+          group.addTask {
+            try await self.requestMaker.makeRequest()
+          }
+        }
+
+        /// While `running` is true, we'll keep launching new requests to
+        /// maintain `maxPermittedOutstandingRequests` running
+        /// at any given time.
+        for try await _ in group {
+          if self.running.load(ordering: .relaxed) {
+            group.addTask {
+              try await self.requestMaker.makeRequest()
+            }
+          }
+        }
+        self.stopIsComplete()
+      }
+    }
+
+    /// Get stats for sending to the driver.
+    /// - parameters:
+    ///     - reset: Should the stats reset after copying.
+    /// - returns: The statistics for this channel.
+    func getStats(reset: Bool) -> Stats {
+      return self.stats.copyData(reset: reset)
+    }
+
+    /// Start sending requests to the server.
+    func start() async throws {
+      try await self.launchRequests()
+    }
+
+    private func stopIsComplete() {
+      // Close the connection then signal done.
+      self.channel.close().cascade(to: self.stopComplete)
+    }
+
+    /// Stop sending requests to the server.
+    /// - returns: A future which can be waited on to signal when all activity has ceased.
+    func stop() async throws {
+      self.requestMaker.requestStop()
+      self.running.store(false, ordering: .relaxed)
+      try await self.stopComplete.futureResult.get()
+    }
+  }
+}
+
+/// Create an asynchronous client of the requested type.
+/// - parameters:
+///     - config: Description of the client required.
+/// - returns: The client created.
+func makeAsyncClient(config: Grpc_Testing_ClientConfig) throws -> AsyncQPSClient {
+  switch config.rpcType {
+  case .unary:
+    return try AsyncQPSClientImpl<AsyncUnaryRequestMaker>(config: config)
+  case .streaming:
+    return try AsyncQPSClientImpl<AsyncPingPongRequestMaker>(config: config)
+  case .streamingFromClient,
+       .streamingFromServer,
+       .streamingBothWays:
+    throw GRPCStatus(code: .unimplemented, message: "Client Type not implemented")
+  case .UNRECOGNIZED:
+    throw GRPCStatus(code: .invalidArgument, message: "Unrecognised client rpc type")
+  }
+}

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

@@ -0,0 +1,102 @@
+/*
+ * 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 Foundation
+import GRPC
+import Logging
+import NIOCore
+import NIOPosix
+
+/// Server setup for asynchronous requests.
+final class AsyncQPSServerImpl: AsyncQPSServer {
+  private let logger = Logger(label: "AsyncQPSServerImpl")
+
+  private let eventLoopGroup: MultiThreadedEventLoopGroup
+  private let server: Server
+  private let threadCount: Int
+
+  private var statsPeriodStart: DispatchTime
+  private var cpuStatsPeriodStart: CPUTime
+
+  var serverInfo: ServerInfo {
+    let port = self.server.channel.localAddress?.port ?? 0
+    return ServerInfo(threadCount: self.threadCount, port: port)
+  }
+
+  /// Initialisation.
+  /// - parameters:
+  ///     - config: Description of the type of server required.
+  init(config: Grpc_Testing_ServerConfig) async throws {
+    // Setup threads as requested.
+    let threadCount = config.asyncServerThreads > 0
+      ? Int(config.asyncServerThreads)
+      : System.coreCount
+    self.threadCount = threadCount
+    self.logger.info("Sizing AsyncQPSServerImpl", metadata: ["threads": "\(threadCount)"])
+    self.eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: threadCount)
+
+    // Start stats gathering.
+    self.statsPeriodStart = grpcTimeNow()
+    self.cpuStatsPeriodStart = getResourceUsage()
+
+    let workerService = AsyncBenchmarkServiceImpl()
+
+    // Start the server
+    self.server = try await Server.insecure(group: self.eventLoopGroup)
+      .withServiceProviders([workerService])
+      .withLogger(self.logger)
+      .bind(host: "localhost", port: Int(config.port))
+      .get()
+  }
+
+  /// 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 which the status should be sent.
+  func sendStatus(
+    reset: Bool,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ServerStatus>
+  ) async throws {
+    let currentTime = grpcTimeNow()
+    let currentResourceUsage = getResourceUsage()
+    var result = Grpc_Testing_ServerStatus()
+    result.stats.timeElapsed = (currentTime - self.statsPeriodStart).asSeconds()
+    result.stats.timeSystem = currentResourceUsage.systemTime - self.cpuStatsPeriodStart
+      .systemTime
+    result.stats.timeUser = currentResourceUsage.userTime - self.cpuStatsPeriodStart.userTime
+    result.stats.totalCpuTime = 0
+    result.stats.idleCpuTime = 0
+    result.stats.cqPollCount = 0
+    self.logger.info("Sending server status")
+    try await responseStream.send(result)
+    if reset {
+      self.statsPeriodStart = currentTime
+      self.cpuStatsPeriodStart = currentResourceUsage
+    }
+  }
+
+  /// Shut down the service.
+  func shutdown() async throws {
+    do {
+      try await self.server.initiateGracefulShutdown().get()
+    } catch {
+      self.logger.error("Error closing server", metadata: ["error": "\(error)"])
+      // May as well plough on anyway -
+      // we will hopefully sort outselves out shutting down the eventloops
+    }
+    try await self.eventLoopGroup.shutdownGracefully()
+  }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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 Logging
+
+/// Implement to provide a method of making requests to a server from a client.
+protocol AsyncRequestMaker: Sendable {
+  /// 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
+  )
+
+  /// Initiate a request sequence to the server.
+  func makeRequest() async throws
+
+  /// Request termination of the request-response sequence.
+  func requestStop()
+}

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/Async/AsyncServerProtocol.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
+
+/// Interface server types must implement when using async APIs.
+protocol AsyncQPSServer {
+  /// The server information for this server.
+  var serverInfo: ServerInfo { get }
+
+  /// 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_ServerStatus>
+  ) async throws
+
+  /// Shut down the service.
+  func shutdown() async throws
+}

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

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2020, 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 Logging
+import NIOCore
+
+/// Makes unary requests to the server and records performance statistics.
+final class AsyncUnaryRequestMaker: AsyncRequestMaker {
+  private let client: Grpc_Testing_BenchmarkServiceAsyncClient
+  private let requestMessage: Grpc_Testing_SimpleRequest
+  private let logger: Logger
+  private let stats: StatsWithLock
+
+  /// 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: Logging.Logger,
+    stats: StatsWithLock
+  ) {
+    self.client = client
+    self.requestMessage = requestMessage
+    self.logger = logger
+    self.stats = stats
+  }
+
+  /// Initiate a request sequence to the server - in this case a single unary requests and wait for a response.
+  /// - returns: A future which completes when the request-response sequence is complete.
+  func makeRequest() async throws {
+    let startTime = grpcTimeNow()
+    do {
+      _ = try await self.client.unaryCall(self.requestMessage)
+      let endTime = grpcTimeNow()
+      self.stats.add(latency: endTime - startTime)
+    } catch {
+      self.logger.error("Error from unary request", metadata: ["error": "\(error)"])
+      throw error
+    }
+  }
+
+  /// Request termination of the request-response sequence.
+  func requestStop() {
+    // No action here - we could potentially try and cancel the request easiest to just wait.
+  }
+}

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

@@ -0,0 +1,335 @@
+/*
+ * 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
+
+// Implementation of the control service for communication with the driver process.
+actor AsyncWorkerServiceImpl: Grpc_Testing_WorkerServiceAsyncProvider {
+  let interceptors: Grpc_Testing_WorkerServiceServerInterceptorFactoryProtocol? = nil
+
+  private let finishedPromise: EventLoopPromise<Void>
+  private let serverPortOverride: Int?
+
+  private var runningServer: AsyncQPSServer?
+  private var runningClient: AsyncQPSClient?
+
+  /// Initialise.
+  /// - parameters:
+  ///     - finishedPromise:  Promise to complete when the server has finished running.
+  ///     - serverPortOverride: An override to port number requested by the driver process.
+  init(finishedPromise: EventLoopPromise<Void>, serverPortOverride: Int?) {
+    self.finishedPromise = finishedPromise
+    self.serverPortOverride = serverPortOverride
+  }
+
+  /// Start server with specified workload.
+  /// First request sent specifies the ServerConfig followed by ServerStatus
+  /// response. After that, a "Mark" can be sent anytime to request the latest
+  /// stats. Closing the stream will initiate shutdown of the test server
+  /// and once the shutdown has finished, the OK status is sent to terminate
+  /// this RPC.
+  func runServer(
+    requestStream: GRPCAsyncRequestStream<Grpc_Testing_ServerArgs>,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ServerStatus>,
+    context: GRPCAsyncServerCallContext
+  ) async throws {
+    context.request.logger.info("runServer stream started.")
+    for try await request in requestStream {
+      try await self.handleServerMessage(
+        context: context,
+        args: request,
+        responseStream: responseStream
+      )
+    }
+    try await self.handleServerEnd(context: context)
+  }
+
+  /// Start client with specified workload.
+  /// First request sent specifies the ClientConfig followed by ClientStatus
+  /// response. After that, a "Mark" can be sent anytime to request the latest
+  /// stats. Closing the stream will initiate shutdown of the test client
+  /// and once the shutdown has finished, the OK status is sent to terminate
+  /// this RPC.
+  func runClient(
+    requestStream: GRPCAsyncRequestStream<Grpc_Testing_ClientArgs>,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ClientStatus>,
+    context: GRPCAsyncServerCallContext
+  ) async throws {
+    for try await request in requestStream {
+      try await self.handleClientMessage(
+        context: context,
+        args: request,
+        responseStream: responseStream
+      )
+    }
+    try await self.handleClientEnd(context: context)
+  }
+
+  /// Just return the core count - unary call
+  func coreCount(
+    request: Grpc_Testing_CoreRequest,
+    context: GRPCAsyncServerCallContext
+  ) async throws -> Grpc_Testing_CoreResponse {
+    context.request.logger.notice("coreCount queried")
+    return Grpc_Testing_CoreResponse.with { $0.cores = Int32(System.coreCount) }
+  }
+
+  /// Quit this worker
+  func quitWorker(
+    request: Grpc_Testing_Void,
+    context: GRPCAsyncServerCallContext
+  ) -> Grpc_Testing_Void {
+    context.request.logger.warning("quitWorker called")
+    self.finishedPromise.succeed(())
+    return Grpc_Testing_Void()
+  }
+
+  // MARK: Run Server
+
+  /// Handle a message received from the driver about operating as a server.
+  private func handleServerMessage(
+    context: GRPCAsyncServerCallContext,
+    args: Grpc_Testing_ServerArgs,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ServerStatus>
+  ) async throws {
+    switch args.argtype {
+    case let .some(.setup(serverConfig)):
+      try await self.handleServerSetup(
+        context: context,
+        config: serverConfig,
+        responseStream: responseStream
+      )
+    case let .some(.mark(mark)):
+      try await self.handleServerMarkRequested(
+        context: context,
+        mark: mark,
+        responseStream: responseStream
+      )
+    case .none:
+      ()
+    }
+  }
+
+  /// Handle a request to setup a server.
+  /// Makes a new server and sets it running.
+  private func handleServerSetup(
+    context: GRPCAsyncServerCallContext,
+    config: Grpc_Testing_ServerConfig,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ServerStatus>
+  ) async throws {
+    context.request.logger.info("server setup requested")
+    guard self.runningServer == nil else {
+      context.request.logger.error("server already running")
+      throw GRPCStatus(
+        code: GRPCStatus.Code.resourceExhausted,
+        message: "Server worker busy"
+      )
+    }
+    try await self.runServerBody(
+      context: context,
+      serverConfig: config,
+      responseStream: responseStream
+    )
+  }
+
+  /// Gathers stats and returns them to the driver process.
+  private func handleServerMarkRequested(
+    context: GRPCAsyncServerCallContext,
+    mark: Grpc_Testing_Mark,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ServerStatus>
+  ) async throws {
+    context.request.logger.info("server mark requested")
+    guard let runningServer = self.runningServer else {
+      context.request.logger.error("server not running")
+      throw GRPCStatus(
+        code: GRPCStatus.Code.failedPrecondition,
+        message: "Server not running"
+      )
+    }
+    try await runningServer.sendStatus(reset: mark.reset, responseStream: responseStream)
+  }
+
+  /// Handle a message from the driver asking this server function to stop running.
+  private func handleServerEnd(context: GRPCAsyncServerCallContext) async throws {
+    context.request.logger.info("runServer stream ended.")
+    if let runningServer = self.runningServer {
+      self.runningServer = nil
+      try await runningServer.shutdown()
+    }
+  }
+
+  // MARK: Create Server
+
+  /// Start a server running of the requested type.
+  private func runServerBody(
+    context: GRPCAsyncServerCallContext,
+    serverConfig: Grpc_Testing_ServerConfig,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ServerStatus>
+  ) async throws {
+    var serverConfig = serverConfig
+    self.serverPortOverride.map { serverConfig.port = Int32($0) }
+
+    self.runningServer = try await AsyncWorkerServiceImpl.createServer(
+      context: context,
+      config: serverConfig,
+      responseStream: responseStream
+    )
+  }
+
+  private static func sendServerInfo(
+    _ serverInfo: ServerInfo,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ServerStatus>
+  ) async throws {
+    var response = Grpc_Testing_ServerStatus()
+    response.cores = Int32(serverInfo.threadCount)
+    response.port = Int32(serverInfo.port)
+    try await responseStream.send(response)
+  }
+
+  /// Create a server of the requested type.
+  private static func createServer(
+    context: GRPCAsyncServerCallContext,
+    config: Grpc_Testing_ServerConfig,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ServerStatus>
+  ) async throws -> AsyncQPSServer {
+    context.request.logger.info(
+      "Starting server",
+      metadata: ["type": .stringConvertible(config.serverType)]
+    )
+
+    switch config.serverType {
+    case .asyncServer:
+      let asyncServer = try await AsyncQPSServerImpl(config: config)
+      let serverInfo = asyncServer.serverInfo
+      try await self.sendServerInfo(serverInfo, responseStream: responseStream)
+      return asyncServer
+    case .syncServer,
+         .asyncGenericServer,
+         .otherServer,
+         .callbackServer:
+      throw GRPCStatus(code: .unimplemented, message: "Server Type not implemented")
+    case .UNRECOGNIZED:
+      throw GRPCStatus(code: .invalidArgument, message: "Unrecognised server type")
+    }
+  }
+
+  // MARK: Run Client
+
+  /// Handle a message from the driver about operating as a client.
+  private func handleClientMessage(
+    context: GRPCAsyncServerCallContext,
+    args: Grpc_Testing_ClientArgs,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ClientStatus>
+  ) async throws {
+    switch args.argtype {
+    case let .some(.setup(clientConfig)):
+      try await self.handleClientSetup(
+        context: context,
+        config: clientConfig,
+        responseStream: responseStream
+      )
+      self.runningClient!.startClient()
+    case let .some(.mark(mark)):
+      // Capture stats
+      try await self.handleClientMarkRequested(
+        context: context,
+        mark: mark,
+        responseStream: responseStream
+      )
+    case .none:
+      ()
+    }
+  }
+
+  /// Setup a client as described by the message from the driver.
+  private func handleClientSetup(
+    context: GRPCAsyncServerCallContext,
+    config: Grpc_Testing_ClientConfig,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ClientStatus>
+  ) async throws {
+    context.request.logger.info("client setup requested")
+    guard self.runningClient == nil else {
+      context.request.logger.error("client already running")
+      throw GRPCStatus(
+        code: GRPCStatus.Code.resourceExhausted,
+        message: "Client worker busy"
+      )
+    }
+    try self.runClientBody(context: context, clientConfig: config)
+    // Initial status is the default (in C++)
+    try await responseStream.send(Grpc_Testing_ClientStatus())
+  }
+
+  /// Captures stats and send back to driver process.
+  private func handleClientMarkRequested(
+    context: GRPCAsyncServerCallContext,
+    mark: Grpc_Testing_Mark,
+    responseStream: GRPCAsyncResponseStreamWriter<Grpc_Testing_ClientStatus>
+  ) async throws {
+    context.request.logger.info("client mark requested")
+    guard let runningClient = self.runningClient else {
+      context.request.logger.error("client not running")
+      throw GRPCStatus(
+        code: GRPCStatus.Code.failedPrecondition,
+        message: "Client not running"
+      )
+    }
+    try await runningClient.sendStatus(reset: mark.reset, responseStream: responseStream)
+  }
+
+  /// Call when an end message has been received.
+  /// Causes the running client to shutdown.
+  private func handleClientEnd(context: GRPCAsyncServerCallContext) async throws {
+    context.request.logger.info("runClient ended")
+    if let runningClient = self.runningClient {
+      self.runningClient = nil
+      try await runningClient.shutdown()
+    }
+  }
+
+  // MARK: Create Client
+
+  /// Setup and run a client of the requested type.
+  private func runClientBody(
+    context: GRPCAsyncServerCallContext,
+    clientConfig: Grpc_Testing_ClientConfig
+  ) throws {
+    self.runningClient = try AsyncWorkerServiceImpl.makeClient(
+      context: context,
+      clientConfig: clientConfig
+    )
+  }
+
+  /// Create a client of the requested type.
+  private static func makeClient(
+    context: GRPCAsyncServerCallContext,
+    clientConfig: Grpc_Testing_ClientConfig
+  ) throws -> AsyncQPSClient {
+    switch clientConfig.clientType {
+    case .asyncClient:
+      if case .bytebufParams = clientConfig.payloadConfig.payload {
+        throw GRPCStatus(code: .unimplemented, message: "Client Type not implemented")
+      }
+      return try makeAsyncClient(config: clientConfig)
+    case .syncClient,
+         .otherClient,
+         .callbackClient:
+      throw GRPCStatus(code: .unimplemented, message: "Client Type not implemented")
+    case .UNRECOGNIZED:
+      throw GRPCStatus(code: .invalidArgument, message: "Unrecognised client type")
+    }
+  }
+}

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/BenchmarkServiceImpl.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIOBenchmarkServiceImpl.swift

@@ -19,7 +19,7 @@ import GRPC
 import NIOCore
 
 /// Implementation of asynchronous service for benchmarking.
-final class AsyncQPSServerImpl: Grpc_Testing_BenchmarkServiceProvider {
+final class NIOBenchmarkServiceImpl: Grpc_Testing_BenchmarkServiceProvider {
   let interceptors: Grpc_Testing_BenchmarkServiceServerInterceptorFactoryProtocol? = nil
 
   /// One request followed by one response.
@@ -30,7 +30,7 @@ final class AsyncQPSServerImpl: Grpc_Testing_BenchmarkServiceProvider {
   ) -> EventLoopFuture<Grpc_Testing_SimpleResponse> {
     do {
       return context.eventLoop
-        .makeSucceededFuture(try AsyncQPSServerImpl.processSimpleRPC(request: request))
+        .makeSucceededFuture(try NIOBenchmarkServiceImpl.processSimpleRPC(request: request))
     } catch {
       return context.eventLoop.makeFailedFuture(error)
     }
@@ -46,7 +46,7 @@ final class AsyncQPSServerImpl: Grpc_Testing_BenchmarkServiceProvider {
       switch event {
       case let .message(request):
         do {
-          let response = try AsyncQPSServerImpl.processSimpleRPC(request: request)
+          let response = try NIOBenchmarkServiceImpl.processSimpleRPC(request: request)
           context.sendResponse(response, promise: nil)
         } catch {
           context.statusPromise.fail(error)

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/ClientInterface.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIOClientProtocol.swift

@@ -18,7 +18,7 @@ import GRPC
 import NIOCore
 
 /// Protocol which clients must implement.
-protocol QPSClient {
+protocol NIOQPSClient {
   /// Send the status of the current test
   /// - parameters:
   ///     - reset: Indicates if the stats collection should be reset after publication or not.

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/AsyncPingPongRequestMaker.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIOPingPongRequestMaker.swift

@@ -21,7 +21,7 @@ import NIOCore
 
 /// Makes streaming requests and listens to responses ping-pong style.
 /// Iterations can be limited by config.
-final class AsyncPingPongRequestMaker: RequestMaker {
+final class NIOPingPongRequestMaker: NIORequestMaker {
   private let client: Grpc_Testing_BenchmarkServiceNIOClient
   private let requestMessage: Grpc_Testing_SimpleRequest
   private let logger: Logger

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/AsyncClient.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIOQPSClientImpl.swift

@@ -24,11 +24,11 @@ import NIOCore
 import NIOPosix
 
 /// Client to make a series of asynchronous calls.
-final class AsyncQPSClient<RequestMakerType: RequestMaker>: QPSClient {
+final class NIOQPSClientImpl<RequestMakerType: NIORequestMaker>: NIOQPSClient {
   private let eventLoopGroup: MultiThreadedEventLoopGroup
   private let threadCount: Int
 
-  private let logger = Logger(label: "AsyncQPSClient")
+  private let logger = Logger(label: "NIOQPSClientImpl")
 
   private let channelRepeaters: [ChannelRepeater]
 
@@ -46,7 +46,7 @@ final class AsyncQPSClient<RequestMakerType: RequestMaker>: QPSClient {
     // Setup threads
     let threadCount = config.threadsToUse()
     self.threadCount = threadCount
-    self.logger.info("Sizing AsyncQPSClient", metadata: ["threads": "\(threadCount)"])
+    self.logger.info("Sizing NIOQPSClientImpl", metadata: ["threads": "\(threadCount)"])
     let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: threadCount)
     self.eventLoopGroup = eventLoopGroup
 
@@ -54,7 +54,7 @@ final class AsyncQPSClient<RequestMakerType: RequestMaker>: QPSClient {
     self.statsPeriodStart = grpcTimeNow()
     self.cpuStatsPeriodStart = getResourceUsage()
 
-    let requestMessage = try AsyncQPSClient
+    let requestMessage = try NIOQPSClientImpl
       .makeClientRequest(payloadConfig: config.payloadConfig)
 
     // Start the requested number of channels.
@@ -276,12 +276,12 @@ final class AsyncQPSClient<RequestMakerType: RequestMaker>: QPSClient {
 /// - parameters:
 ///     - config: Description of the client required.
 /// - returns: The client created.
-func makeAsyncClient(config: Grpc_Testing_ClientConfig) throws -> QPSClient {
+func makeAsyncClient(config: Grpc_Testing_ClientConfig) throws -> NIOQPSClient {
   switch config.rpcType {
   case .unary:
-    return try AsyncQPSClient<AsyncUnaryRequestMaker>(config: config)
+    return try NIOQPSClientImpl<NIOUnaryRequestMaker>(config: config)
   case .streaming:
-    return try AsyncQPSClient<AsyncPingPongRequestMaker>(config: config)
+    return try NIOQPSClientImpl<NIOPingPongRequestMaker>(config: config)
   case .streamingFromClient:
     throw GRPCStatus(code: .unimplemented, message: "Client Type not implemented")
   case .streamingFromServer:

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/AsyncServer.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIOQPSServerImpl.swift

@@ -20,8 +20,8 @@ import Logging
 import NIOCore
 import NIOPosix
 
-/// Server setup for asynchronous requests.
-final class AsyncQPSServer: QPSServer {
+/// Server setup for asynchronous requests (using EventLoopFutures).
+final class NIOQPSServerImpl: NIOQPSServer {
   private let eventLoopGroup: MultiThreadedEventLoopGroup
   private let server: EventLoopFuture<Server>
   private let threadCount: Int
@@ -48,7 +48,7 @@ final class AsyncQPSServer: QPSServer {
     self.statsPeriodStart = grpcTimeNow()
     self.cpuStatsPeriodStart = getResourceUsage()
 
-    let workerService = AsyncQPSServerImpl()
+    let workerService = NIOBenchmarkServiceImpl()
 
     // Start the server.
     // TODO: Support TLS if requested.

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/RequestMaker.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIORequestMaker.swift

@@ -19,7 +19,7 @@ import Logging
 import NIOCore
 
 /// Implement to provide a method of making requests to a server from a client.
-protocol RequestMaker {
+protocol NIORequestMaker {
   /// Initialiser to gather requirements.
   /// - Parameters:
   ///    - config: config from the driver describing what to do.

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/ServerInterface.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIOServerProtocol.swift

@@ -17,8 +17,8 @@
 import GRPC
 import NIOCore
 
-/// Interface server types must implement.
-protocol QPSServer {
+/// Interface server types must implement when using NIO.
+protocol NIOQPSServer {
   /// Send the status of the current test
   /// - parameters:
   ///     - reset: Indicates if the stats collection should be reset after publication or not.

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/AsyncUnaryRequestMaker.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIOUnaryRequestMaker.swift

@@ -19,7 +19,7 @@ import Logging
 import NIOCore
 
 /// Makes unary requests to the server and records performance statistics.
-final class AsyncUnaryRequestMaker: RequestMaker {
+final class NIOUnaryRequestMaker: NIORequestMaker {
   private let client: Grpc_Testing_BenchmarkServiceNIOClient
   private let requestMessage: Grpc_Testing_SimpleRequest
   private let logger: Logger

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/WorkerServiceImpl.swift → Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/NIO/NIOWorkerServiceImpl.swift

@@ -18,14 +18,14 @@ import GRPC
 import NIOCore
 
 // Implementation of the control service for communication with the driver process.
-class WorkerServiceImpl: Grpc_Testing_WorkerServiceProvider {
+class NIOWorkerServiceImpl: Grpc_Testing_WorkerServiceProvider {
   let interceptors: Grpc_Testing_WorkerServiceServerInterceptorFactoryProtocol? = nil
 
   private let finishedPromise: EventLoopPromise<Void>
   private let serverPortOverride: Int?
 
-  private var runningServer: QPSServer?
-  private var runningClient: QPSClient?
+  private var runningServer: NIOQPSServer?
+  private var runningClient: NIOQPSClient?
 
   /// Initialise.
   /// - parameters:
@@ -173,7 +173,7 @@ class WorkerServiceImpl: Grpc_Testing_WorkerServiceProvider {
     self.serverPortOverride.map { serverConfig.port = Int32($0) }
 
     do {
-      self.runningServer = try WorkerServiceImpl.createServer(
+      self.runningServer = try NIOWorkerServiceImpl.createServer(
         context: context,
         config: serverConfig
       )
@@ -186,7 +186,7 @@ class WorkerServiceImpl: Grpc_Testing_WorkerServiceProvider {
   private static func createServer(
     context: StreamingResponseCallContext<Grpc_Testing_ServerStatus>,
     config: Grpc_Testing_ServerConfig
-  ) throws -> QPSServer {
+  ) throws -> NIOQPSServer {
     context.logger.info(
       "Starting server",
       metadata: ["type": .stringConvertible(config.serverType)]
@@ -196,7 +196,7 @@ class WorkerServiceImpl: Grpc_Testing_WorkerServiceProvider {
     case .syncServer:
       throw GRPCStatus(code: .unimplemented, message: "Server Type not implemented")
     case .asyncServer:
-      let asyncServer = AsyncQPSServer(
+      let asyncServer = NIOQPSServerImpl(
         config: config,
         whenBound: { serverInfo in
           var response = Grpc_Testing_ServerStatus()
@@ -297,7 +297,7 @@ class WorkerServiceImpl: Grpc_Testing_WorkerServiceProvider {
     clientConfig: Grpc_Testing_ClientConfig
   ) {
     do {
-      self.runningClient = try WorkerServiceImpl.makeClient(
+      self.runningClient = try NIOWorkerServiceImpl.makeClient(
         context: context,
         clientConfig: clientConfig
       )
@@ -310,7 +310,7 @@ class WorkerServiceImpl: Grpc_Testing_WorkerServiceProvider {
   private static func makeClient(
     context: StreamingResponseCallContext<Grpc_Testing_ClientStatus>,
     clientConfig: Grpc_Testing_ClientConfig
-  ) throws -> QPSClient {
+  ) throws -> NIOQPSClient {
     switch clientConfig.clientType {
     case .syncClient:
       throw GRPCStatus(code: .unimplemented, message: "Client Type not implemented")

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/QPSWorker.swift

@@ -22,16 +22,18 @@ import NIOPosix
 /// Sets up and runs a worker service which listens for instructions on what tests to run.
 /// Currently doesn't understand TLS for communication with the driver.
 class QPSWorker {
-  private var driverPort: Int
-  private var serverPort: Int?
+  private let driverPort: Int
+  private let serverPort: Int?
+  private let useAsync: Bool
 
   /// Initialise.
   /// - parameters:
   ///     - driverPort: Port to listen for instructions on.
   ///     - serverPort: Possible override for the port the testing will actually occur on - usually supplied by the driver process.
-  init(driverPort: Int, serverPort: Int?) {
+  init(driverPort: Int, serverPort: Int?, useAsync: Bool) {
     self.driverPort = driverPort
     self.serverPort = serverPort
+    self.useAsync = useAsync
   }
 
   private let logger = Logger(label: "QPSWorker")
@@ -49,12 +51,20 @@ class QPSWorker {
     let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
     self.eventLoopGroup = eventLoopGroup
 
+    let workerService: CallHandlerProvider
     let workEndPromise: EventLoopPromise<Void> = eventLoopGroup.next().makePromise()
     workEndPromise.futureResult.whenSuccess(onQuit)
-    let workerService = WorkerServiceImpl(
-      finishedPromise: workEndPromise,
-      serverPortOverride: self.serverPort
-    )
+    if self.useAsync {
+      workerService = AsyncWorkerServiceImpl(
+        finishedPromise: workEndPromise,
+        serverPortOverride: self.serverPort
+      )
+    } else {
+      workerService = NIOWorkerServiceImpl(
+        finishedPromise: workEndPromise,
+        serverPortOverride: self.serverPort
+      )
+    }
 
     // Start the server.
     self.logger.info("Binding to localhost", metadata: ["driverPort": "\(self.driverPort)"])

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/Stats.swift

@@ -28,7 +28,8 @@ struct Stats {
 /// Stats with access controlled by a lock -
 /// Needs locking rather than event loop hopping as the driver refuses to wait shutting
 /// the connection immediately after the request.
-class StatsWithLock {
+/// Marked `@unchecked Sendable` since we control access to `data` via a Lock.
+final class StatsWithLock: @unchecked Sendable {
   private var data = Stats()
   private let lock = Lock()
 

Performance/QPSBenchmark/Sources/QPSBenchmark/Runtime/main.swift

@@ -26,11 +26,8 @@ final class QPSWorkerApp: ParsableCommand {
   @Option(name: .customLong("server_port"), help: "Port for operation as a server.")
   var serverPort: Int?
 
-  @Option(
-    name: .customLong("credential_type"),
-    help: "Credential type for communication with driver."
-  )
-  var credentialType: String = "todo" // TODO: Default to kInsecureCredentialsType
+  @Flag
+  var useAsync: Bool = false
 
   /// Run the application and wait for completion to be signalled.
   func run() throws {
@@ -47,11 +44,13 @@ final class QPSWorkerApp: ParsableCommand {
     // This installs backtrace.
     let lifecycle = ServiceLifecycle()
 
+    logger.info("Initializing QPSWorker - useAsync: \(self.useAsync)")
     let qpsWorker = QPSWorker(
       driverPort: self.driverPort,
-      serverPort: self.serverPort
+      serverPort: self.serverPort,
+      useAsync: self.useAsync
     )
-    // credentialType: self.credentialType)
+
     qpsWorker.start {
       lifecycle.shutdown()
     }