QPS Benchmark Service implementation #1828

stefanadranca · 2024-03-13T17:33:19Z

Motivation:

The 2 workers used in QPS testing are a Benchmark service client and server respectivelly, so we need to implement the Benchmark Service.

Modifications:

Implemented the 'BenchmarkService' struct that defines the service protocol methods, based on their documentation.

Result:

We will be able to proceed with the Wrorker Service implementation.

Motivation: The 2 workers used in QPS testing are a Benchmark service client and server respectivelly, so we need to implement the Benchmark Service. Modifications: Implemented the 'BenchmarkService' struct that defines the service protocol methods, based on their documentation. Result: We will be able to proceed with the Wrorker Service implementation.

glbrntt · 2024-03-14T10:30:09Z

Sources/performance-worker/BenchmarkService.swift

+  {
+
+    // Throw an error if the status is not `ok`. Otherwise, an `ok` status is automatically sent


Suggested change

{

// Throw an error if the status is not `ok`. Otherwise, an `ok` status is automatically sent

{

// Throw an error if the status is not `ok`. Otherwise, an `ok` status is automatically sent

glbrntt · 2024-03-14T10:33:55Z

Sources/performance-worker/BenchmarkService.swift

+
+@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+extension BenchmarkService {
+  private func echoStatus(responseStatus: Grpc_Testing_EchoStatus) throws {


The naming/usage is not all that clear because it's try echoStatus(...) which makes it seem like it throws if it can't echo the status.

One analogue I can think of is throwing a cancellation error if a task is cancelled and that's spelled try Task.checkCancellation().

Drawing on that I think something like try checkOkStatus(_ responseStatus: Grpc_Testing_EchoStatus) (note the dropped label as 'status' is included in the name) has clearer semantics: it returns if the status is okay, otherwise it throws.

glbrntt · 2024-03-14T10:35:42Z

Sources/performance-worker/BenchmarkService.swift

+  ) async throws
+    -> ServerResponse.Single<Grpc_Testing_BenchmarkService.Method.StreamingFromClient.Output>
+  {
+    throw RPCError(code: .unimplemented, message: "The RPC is not implemented.")


Why don't we implement this one?

glbrntt · 2024-03-14T10:39:57Z

Sources/performance-worker/BenchmarkService.swift

+            $0.payload = request.message.payload
+          }
+        )
+        try await Task.sleep(nanoseconds: 10)


We shouldn't need to sleep here

glbrntt · 2024-03-14T10:41:34Z

Sources/performance-worker/BenchmarkService.swift

+        try self.echoStatus(responseStatus: request.message.responseStatus)
+      }
+
+      while true {


We need to stop eventually, while !Task.isCancelled is probably a better condition to use here.

glbrntt · 2024-03-14T10:41:50Z

Sources/performance-worker/BenchmarkService.swift

+    -> GRPCCore.ServerResponse.Stream<Grpc_Testing_BenchmarkService.Method.StreamingBothWays.Output>
+  {
+    return ServerResponse.Stream { writer in
+      while true {


Same here, we need to stop eventually

glbrntt · 2024-03-14T10:41:55Z

Sources/performance-worker/BenchmarkService.swift

+            }
+          }
+        )
+        try await Task.sleep(nanoseconds: 10)


No need to sleep here

glbrntt · 2024-03-14T10:42:52Z

Sources/performance-worker/BenchmarkService.swift

+      message: Grpc_Testing_BenchmarkService.Method.UnaryCall.Output.with {
+        $0.payload = request.message.payload
+      }


This isn't quite right: the request tells us how big the response payload should be (via responeSize). Same goes for all the other RPCs.

is this also true for streamingBothWays()? The documentation for it says "Both sides send the content of their own choice to the other."

Yes, I believe so, I think the docs are outdated because the scenario descriptions which drive the tests expect different request/response sizes.

Wouldn't making this change mean that streamingCall() and streamingBothWays() become the same function?

glbrntt · 2024-03-14T10:43:26Z

Sources/performance-worker/BenchmarkService.swift

+    guard let code = Status.Code(rawValue: Int(responseStatus.code))
+    else {


Suggested change

guard let code = Status.Code(rawValue: Int(responseStatus.code))

else {

guard let code = Status.Code(rawValue: Int(responseStatus.code)) else {

glbrntt · 2024-03-14T10:45:14Z

Sources/performance-worker/BenchmarkService.swift

+    let status = Status(code: code, message: responseStatus.message)
+    if let error = RPCError(status: status) {
+      throw error
+    }


nit: we always create the Status but don't always use it (in fact we don't need to use it at all):

if let code = RPCError.Code(code) { throw RPCError(code: code, message: "...") }

glbrntt · 2024-03-15T11:22:45Z

Sources/performance-worker/BenchmarkService.swift

+
+@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+struct BenchmarkService: Grpc_Testing_BenchmarkService.ServiceProtocol {
+  /// Used to check if


The suspense is killing me, what is it used to check?

glbrntt · 2024-03-15T11:25:06Z

Sources/performance-worker/BenchmarkService.swift

+@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+struct BenchmarkService: Grpc_Testing_BenchmarkService.ServiceProtocol {
+  /// Used to check if
+  var working = ManagedAtomic<Bool>(true)


This should be private and a let

glbrntt · 2024-03-15T11:28:18Z

Sources/performance-worker/BenchmarkService.swift

+        try await writer.write(
+          Grpc_Testing_BenchmarkService.Method.StreamingCall.Output.with {
+            $0.payload = Grpc_Testing_Payload.with {
+              $0.body = Data(count: Int(request.message.responseSize))


This will create the message each time which is potentially expensive. Can we create it once before the loop?

glbrntt · 2024-03-15T11:29:24Z

Sources/performance-worker/BenchmarkService.swift

+        try await writer.write(
+          Grpc_Testing_BenchmarkService.Method.StreamingCall.Output.with {
+            $0.payload = Grpc_Testing_Payload.with {
+              $0.body = Data(count: 100)


Can you add a comment about why we used 100? There's a very low probability we'll remember why otherwise

Sorry, was there a specific reason we used exactly 100? (We are using a single value so we can create the message once - as you pointed before I will create it before the loop, but was there a reason for this size specifically?)

https://github.com/grpc/grpc-java/blob/36e9f0dfacc853b5d54f64bb14305f0b4c323589/benchmarks/src/main/java/io/grpc/benchmarks/qps/AsyncServer.java#L192-L199

so because the java implementation does this? my understanding of the comment is that they are using the same size for all responses because the spec allows this, but not that the spec says it should be 100. Am I missing something?

You're not missing anything. We're doing it because Java (and other impls) are using 100, i.e. there is precedent elsewhere and we haven't just picked 100 out of thin air (someone else did, we're just following).

glbrntt · 2024-03-15T15:49:25Z

Sources/performance-worker/BenchmarkService.swift

+    for try await message in request.messages {
+      if message.responseStatus.isInitialized {
+        try self.checkOkStatus(message.responseStatus)
+      }
+    }
+
+    // Always use the same canned response for bidirectional streaming.
+    // This is allowed by the spec.
+    let response = Grpc_Testing_BenchmarkService.Method.StreamingCall.Output.with {
+      $0.payload = Grpc_Testing_Payload.with {
+        $0.body = Data(count: 100)
+      }
+    }
+    return ServerResponse.Stream { writer in
+      while working.load(ordering: .acquiring) {
+        try await writer.write(response)
+      }
+      return [:]
+    }


We need to read and write at the same time which means we need to do the work in a task group within the body of the response stream. We can have one task for reading and one for writing. You can use an atomic bool as the stopping condition for writing (we should stop writing when all requests have been consumed).

glbrntt · 2024-03-18T08:32:39Z

Sources/performance-worker/BenchmarkService.swift

+            try self.checkOkStatus(message.responseStatus)
+          }
+        }
+        _ = self.working.exchange(false, ordering: .acquiring)


I don't think this is quite right: self.working will be toggled by the quit method on the worker service, it's an external trigger to shutdown. Looking at it in a different way: I don't think it makes sense that the end of the request stream for this method would stop the response stream from streaming-from-server.

Instead I think we want a separate atomic to signal between the request stream and response stream.

Thank you! It makes sense to use a different atomic for this. One question though: I should still check when writing the responses if 'working' is set to 'true', besides checking the new atomic, right?

glbrntt · 2024-03-18T08:42:11Z

Sources/performance-worker/BenchmarkService.swift

+      }
+      group.cancelAll()
+    }
+    return serverResponse


This is a bit inverted. At the moment the behaviour will be:

RPC starts

Task group is created

One task runs to consume the request messages

The other task returns the response stream immediately

We then consume the response stream, the first result we'll get will be the server response because it's returned immediately so we cancel the task group (and stop consuming requests)

We then wait for all child tasks in the task group to finish (this should be fast because one has already finished and the other has been cancelled)

Then we return the server response so that gRPC can run it.

Instead we should be consuming the request stream within a task group inside the response stream. The rough shape is something like this:

return ServerResponse.Stream { writer in try await withThrowingTaskGroup(of: Void.self) { group in group.addTask { // consume requests } group.addTask { // write responses } } return [:] }

glbrntt

one nit but looks good otherwise

glbrntt · 2024-03-18T13:36:16Z

Sources/performance-worker/BenchmarkService.swift

+              try self.checkOkStatus(message.responseStatus)
+            }
+          }
+          _ = inboundStreaming.exchange(false, ordering: .acquiring)


No need to exchange, store is fine. We can also use relaxed ordering here (and while loading).

stefanadranca requested review from gjcairo and glbrntt March 13, 2024 17:33

glbrntt requested changes Mar 14, 2024

View reviewed changes

stefanadranca added 3 commits March 14, 2024 14:39

implemented feedback

56edde9

fixed streamingbothways

97defe4

fixed documentation

2a9fd9f

stefanadranca added the version/v2 Relates to v2 label Mar 15, 2024

stefanadranca requested a review from glbrntt March 15, 2024 11:10

Merge branch 'main' into sd-qps-benchmark-service

c94afc0

glbrntt requested changes Mar 15, 2024

View reviewed changes

implemented feedback

9e3748d

stefanadranca requested a review from glbrntt March 15, 2024 14:33

changed comment

67b2c35

glbrntt reviewed Mar 15, 2024

View reviewed changes

added taskgroup implementation to streamingBothWays()

c45645d

stefanadranca requested a review from glbrntt March 15, 2024 18:52

glbrntt requested changes Mar 18, 2024

View reviewed changes

fixed task grpup

ebcc7f2

stefanadranca requested a review from glbrntt March 18, 2024 11:10

glbrntt requested changes Mar 18, 2024

View reviewed changes

stefanadranca and others added 2 commits March 18, 2024 14:33

changed exchange to store

7f71ef3

Merge branch 'main' into sd-qps-benchmark-service

d3bd260

stefanadranca requested a review from glbrntt March 18, 2024 14:39

glbrntt approved these changes Mar 18, 2024

View reviewed changes

glbrntt enabled auto-merge (squash) March 18, 2024 14:43

glbrntt merged commit 62653ba into grpc:main Mar 18, 2024

gjcairo added semver/none No version bump required. and removed semver/none No version bump required. labels Apr 2, 2024

		{

		// Throw an error if the status is not `ok`. Otherwise, an `ok` status is automatically sent

		guard let code = Status.Code(rawValue: Int(responseStatus.code))
		else {

QPS Benchmark Service implementation #1828

QPS Benchmark Service implementation #1828

Uh oh!

Conversation

stefanadranca commented Mar 13, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

glbrntt left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants