Reimplement light server runtime by naoh87 · Pull Request #496 · typelevel/fs2-grpc

naoh87 · 2022-02-05T13:01:59Z

This runtime call to 1 unsafeRun* / RPC.

This PR come from #493

Background #386

naoh87 · 2022-02-05T17:27:27Z

+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package fs2.grpc.internal


Place implement to internal package for binary compat control.
I think Fs2ServerCallHandler I/F is what we really need to care about.
Is this right?

Yes, that is exposed to codegen - everything else should be package private.

naoh87 · 2022-02-05T20:09:55Z

Benchmark

I benchmarked with ghz --cpus=3 -z 60s ..., and
benchmark scinario is https://github.com/LesnyRumcajs/grpc_bench

This PR:        11075 rps
fs2-grpc 2.4.4:  9129 rps
Akka-gRPC:      11772 rps

This PR reduce 75% performance difference from Akka-gRPC

This PR

Summary:
  Count:	664576
  Total:	60.01 s
  Slowest:	41.06 ms
  Fastest:	0.17 ms
  Average:	2.33 ms
  Requests/sec:	11075.27

Latency distribution:
  10 % in 1.00 ms
  25 % in 1.43 ms
  50 % in 1.99 ms
  75 % in 2.75 ms
  90 % in 4.26 ms
  95 % in 5.30 ms
  99 % in 6.89 ms

fs2-grpc 2.4.4

Summary:
  Count:	547798
  Total:	60.00 s
  Slowest:	40.29 ms
  Fastest:	0.26 ms
  Average:	2.84 ms
  Requests/sec:	9129.40

Latency distribution:
  10 % in 1.29 ms
  25 % in 1.84 ms
  50 % in 2.53 ms
  75 % in 3.38 ms
  90 % in 4.89 ms
  95 % in 6.03 ms
  99 % in 7.88 ms

Akka-gRPC

Summary:
  Count:	706347
  Total:	60.00 s
  Slowest:	33.03 ms
  Fastest:	0.13 ms
  Average:	2.14 ms
  Requests/sec:	11772.13

Latency distribution:
  10 % in 0.89 ms
  25 % in 1.28 ms
  50 % in 1.80 ms
  75 % in 2.51 ms
  90 % in 4.04 ms
  95 % in 5.02 ms
  99 % in 6.57 ms

ahjohannessen · 2022-02-05T20:27:31Z

I think the usage of unsafe code is a bit too much. The general mantra for fs2-grpc is to have code that is easy to reason about. I think you can come very far using constructs like Ref and SyncIO and eliminate usage of var. This would make it easier to maintain in the longer run. I know that some parts need change, like sync usage of Dispatcher as it causes context shift. That seems doable with SyncIO and using non-blocking methods on Dispatcher. It would be interesting to see if you could adjust your code to something similar to what is in #486.
I do think that performance is important, but not at all costs when we can get numbers that are close to grpc-java using building blocks from cats-effect and fs2.

ahjohannessen · 2022-02-05T21:50:15Z

This library provides some unsafe constructs: https://github.com/davenverse/condemned/blob/main/core/src/main/scala/io/chrisdavenport/condemned/UnsafeDeferred.scala - But I think most of SyncIO + non-blocking methods on dispatcher should get us a long way, the cats-effect constructs are also under more eyes and have higher test coverage.

ahjohannessen · 2022-02-06T09:45:56Z

@naoh87 Some ideas in condemned seem to have led to this

naoh87 · 2022-02-06T10:05:21Z

+  private def mkListener[Request, Response](
+      run: Request => SyncIO[Cancel],
+      call: Fs2StatefulServerCall[Request, Response],
+      state: Ref[SyncIO, State[Request]]


I replaced var to Ref for your advice.

ahjohannessen · 2022-02-06T10:25:09Z

+        state.get.flatMap(_.cancel.getOrElse(SyncIO.unit)).unsafeRunSync()
+
+      override def onMessage(message: Request): Unit =
+        state.get


access instead of get and use modify from it

I think it's not necessary to use access.
Because all ServerCall.Listener methods called simultaneously.

You mean synchroniously?

Yes. Iam sorry about failed to choose English word.

The caller is free to call an instance from multiple threads, but only one call simultaneously

https://github.com/grpc/grpc-java/blob/v1.44.0/api/src/main/java/io/grpc/ServerCall.java#L42-L56

ahjohannessen · 2022-02-06T10:34:24Z

+import UnsafeChannel.State
+import scala.collection.immutable.Queue
+
+final class UnsafeChannel[A] extends AtomicReference[State[A]](State.Consumed) {


Can you use Ref to implement the state machine?

ahjohannessen · 2022-02-06T16:07:49Z

+import scala.collection.immutable.Queue
+
+@nowarn
+final class UnsafeChannel[A](val ref: Ref[SyncIO, State[A]]) extends AnyVal {


I wonder if you know about https://github.com/typelevel/fs2/blob/main/core/shared/src/main/scala/fs2/concurrent/Channel.scala

This is also interesting: https://github.com/davenverse/process/blob/main/core/shared/src/main/scala/io/chrisdavenport/process/structures/UnsafeByteQueue.scala

Seems like some building blocks for unsafe interop are missing in cats-effect

I know that.
Methods name comes from fs2 one.
I made this parts for removing unnecessary context shift.

ahjohannessen · 2022-02-06T16:29:51Z

It would probably be a good idea to do a PR that only focuses on unaryTo as the other scenarios have more moving parts and more involved.

naoh87 · 2022-02-08T18:20:58Z

@ahjohannessen I was wondering if you have had checked the change.

ahjohannessen · 2022-02-08T18:52:35Z

@naoh87 I have not had time yet, will look soon

ahjohannessen · 2022-02-08T19:29:50Z

+    }
+}
+
+final class Fs2StatefulServerCall[Request, Response](


I think it is a bit confusing to have this and Fs2ServerCall

ahjohannessen · 2022-02-08T19:46:31Z

I need a computer in order to do a review :) on my phone now.

ahjohannessen · 2022-02-09T19:56:54Z

Starting to look good :) I have not yet had time to study the code. I wonder what kind of benchmarks and flamegraphs this has compared to main and grpc-java.

ahjohannessen · 2022-02-10T15:01:51Z

+
+  import Fs2ServerCall.Cancel
+
+  def stream[F[_]](response: Stream[F, Response], dispatcher: Dispatcher[F])(implicit F: Async[F]): SyncIO[Cancel] =


Suggested change

def stream[F[_]](response: Stream[F, Response], dispatcher: Dispatcher[F])(implicit F: Async[F]): SyncIO[Cancel] =

def stream[F[_]](response: Stream[F, Response], dispatcher: Dispatcher[F])(implicit F: Sync[F]): SyncIO[Cancel] =

ahjohannessen · 2022-02-10T15:02:05Z

+      dispatcher
+    )
+
+  def unary[F[_]](response: F[Response], dispatcher: Dispatcher[F])(implicit F: Async[F]): SyncIO[Cancel] =


Suggested change

def unary[F[_]](response: F[Response], dispatcher: Dispatcher[F])(implicit F: Async[F]): SyncIO[Cancel] =

def unary[F[_]](response: F[Response], dispatcher: Dispatcher[F])(implicit F: Sync[F]): SyncIO[Cancel] =

ahjohannessen · 2022-02-10T15:02:33Z

+  }
+
+  private def closeStreamF[F[_]](status: Status, metadata: Metadata)(implicit F: Sync[F]): F[Unit] =
+    F.delay(call.close(status, metadata))


Suggested change

F.delay(call.close(status, metadata))

close(status, metadata).to[F]

ahjohannessen · 2022-02-10T15:09:43Z

+  def close(status: Status, metadata: Metadata): SyncIO[Unit] =
+    SyncIO(call.close(status, metadata))
+
+  private def run[F[_]](completed: F[Unit], dispatcher: Dispatcher[F])(implicit F: Sync[F]): SyncIO[Cancel] = {


Perhaps define a local helper inside run to improve readability:

def handleError(t: Throwable): F[Unit] = t match { case ex: StatusException => closeStreamF(ex.getStatus, Option(ex.getTrailers).getOrElse(new Metadata())) case ex: StatusRuntimeException => closeStreamF(ex.getStatus, Option(ex.getTrailers).getOrElse(new Metadata())) case ex => closeStreamF(Status.INTERNAL.withDescription(ex.getMessage).withCause(ex), new Metadata()) }

Then:

case Outcome.Errored(e) => handleError(e)

ahjohannessen · 2022-02-10T15:10:31Z

+      })
+      SyncIO {
+        cancel()
+        ()


Alternative: SyncIO(cancel()).void

ahjohannessen · 2022-02-10T15:15:36Z

+
+    }
+
+  def unary[F[_]: Async, Request, Response](


Suggested change

def unary[F[_]: Async, Request, Response](

def unary[F[_]: Sync, Request, Response](

ahjohannessen · 2022-02-10T15:15:50Z

+        startCallSync(call, opt)(call => req => call.unary(impl(req, headers), dispatcher)).unsafeRunSync()
+    }
+
+  def stream[F[_]: Async, Request, Response](


Suggested change

def stream[F[_]: Async, Request, Response](

def stream[F[_]: Sync, Request, Response](

ahjohannessen · 2022-02-10T15:16:49Z

+
+package fs2.grpc.internal.server
+
+import cats.effect.Async


Suggested change

import cats.effect.Async

import cats.effect.Sync

ahjohannessen · 2022-02-10T15:17:31Z

+  )(f: Fs2ServerCall[Request, Response] => Request => SyncIO[Cancel]): SyncIO[ServerCall.Listener[Request]] = {
+    for {
+      call <- Fs2ServerCall.setup(options, call)
+      _ <- call.request(2)


Perhaps a comment to why two requests are done.

I know this is what grpc-java does:

// We expect only 1 request, but we ask for 2 requests here so that if a misbehaving client // sends more than 1 requests, ServerCall will catch it...

from ServerCalls - A future maintainer would be happy to learn why this is.

ahjohannessen · 2022-02-10T15:27:47Z

+import io.grpc._
+import fs2._
+
+object Fs2ServerCall {


Suggested change

object Fs2ServerCall {

private[server] object Fs2ServerCall {

ahjohannessen · 2022-02-10T15:29:31Z

+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package fs2.grpc.internal.server


Move internal in under server.

ahjohannessen · 2022-02-10T15:29:46Z

+    }
+}
+
+final class Fs2ServerCall[Request, Response](


Suggested change

final class Fs2ServerCall[Request, Response](

private[server] final class Fs2ServerCall[Request, Response](

ahjohannessen · 2022-02-10T15:30:14Z

+import fs2.grpc.server.ServerOptions
+import io.grpc._
+
+object Fs2UnaryServerCallHandler {


Suggested change

object Fs2UnaryServerCallHandler {

private[server] object Fs2UnaryServerCallHandler {

ahjohannessen · 2022-02-10T15:37:15Z

+    run(
+      response.pull.peek1
+        .flatMap {
+          case Some((_, tail)) =>


Calling this tail implies that head of response is not in there, which it is tl.cons(hd)), i.e.:

uncons.flatMap { case None => Pull.pure(None) case Some((hd, tl)) => Pull.pure(Some((hd(0), tl.cons(hd)))) }

Suggestion: rename tail to stream or similar.

This looks good.

naoh87 · 2022-02-11T04:22:59Z

async-profiler results

naoh87 · 2022-02-13T12:30:28Z

Unary Benchmark

1-cpu time restricted server benchmark on Linux(kernel 5.4.0).
I run ghq from MBP with ghq --cpu=4 --connections=5 ...

#496

actual cpu used: 0.90 core
Summary:
  Count:	647800
  Total:	60.00 s
  Slowest:	37.50 ms
  Fastest:	0.24 ms
  Average:	2.12 ms
  Requests/sec:	10796.08

v2.4.4

actual cpu used: 1.0 core
Summary:
  Count:	315765
  Total:	60.00 s
  Slowest:	73.58 ms
  Fastest:	0.30 ms
  Average:	7.65 ms
  Requests/sec:	5262.45

ScalaPB

actual cpu used: 0.75 core
Summary:
  Count:	644442
  Total:	60.00 s
  Slowest:	30.59 ms
  Fastest:	0.23 ms
  Average:	2.12 ms
  Requests/sec:	10740.29

ServerStreaming Benchmark

Benchmark with same condition with unary benchmark
fs2 implement

Stream.emit(HelloReply(request.request)).repeatN(100)

ScalaPB implement

(0 until 100).foreach(_ => responseObserver.onNext(HelloReply(request.request)))
responseObserver.onCompleted()

#496

actual cpu used: 1.0 core
Summary:
  Count:	79699
  Total:	60.00 s
  Slowest:	104.91 ms
  Fastest:	1.54 ms
  Average:	33.53 ms
  Requests/sec:	1328.28

v2.4.4

actual cpu used: 1.0 core
Summary:
  Count:	72239
  Total:	60.00 s
  Slowest:	120.49 ms
  Fastest:	1.92 ms
  Average:	37.31 ms
  Requests/sec:	1203.94

ScalaPB

actual cpu used: 0.85 core
Summary:
  Count:	131386
  Total:	60.01 s
  Slowest:	133.41 ms
  Fastest:	1.71 ms
  Average:	16.36 ms
  Requests/sec:	2189.54

ahjohannessen · 2022-02-13T15:56:44Z

+import fs2.grpc.server.ServerCallOptions
+import io.grpc._
+
+private[server] object Fs2ServerCall {


Perhaps place this inside the handler as it is only a helper.

I use this for streaming handler after this PR.

ahjohannessen · 2022-02-13T16:01:31Z

+    def init[A](cb: A => SyncIO[Cancel]): SyncIO[Ref[SyncIO, Context[A]]] =
+      Ref[SyncIO].of[Context[A]](BeforeCall(cb, None))
+  }
+  case class BeforeCall[A](cb: A => SyncIO[Cancel], request: Option[A]) extends Context[A] {


This name is a bit hard to understand when looking at the fields.

How about BeforeCall(callback: ..., received: ...) ?

Would it make sense to split that into two cases?

Sorry. I don't speak English well.
I couldn't remember what two case indicate.
Could you mind tell me another easy explanation?

Two cases express A => SyncIO[Cancel] will not called after Called if two cases means BeforeCall and Called

Not sure if it makes sense to split it up more, but the cases are at least

cancelled

request message pending

request message received and half-closed pending

request message received and half-closed received

call-completed

…

I think there is a few state to hold over Listener methods.

Is that mean it is more better split like BeforeReceive, Received, Called?
It looks good.

Yes

I think there is a few state to hold over Listener methods.

Not sure what you mean by that

Ref is used by saving state over ServerCall.Listener method.
I thought this doesn't introduce a bunch of state.

ahjohannessen · 2022-02-13T16:02:52Z

+
+  import Fs2ServerCall.Cancel
+
+  sealed trait Context[A]


Is this not more like a call status? I wonder if more explicit cases would make the code easier to reason about

How do you think CallerState?

That is more descriptive 👍

ahjohannessen · 2022-02-13T20:11:02Z

I think we are good, if you rebase and squash I’ll merge. Do the extra states have a measurable impact on req/s ?

naoh87 · 2022-02-13T20:40:22Z

I rebased to one commit and theres was no measurable performance impact.

ahjohannessen · 2022-02-13T20:44:16Z

+      // We expect only 1 request, but we ask for 2 requests here so that if a misbehaving client
+      // sends more than 1 requests, ServerCall will catch it.
+      _ <- call.request(2)
+      ctx <- CallerState.init(f(call))


ahjohannessen · 2022-02-13T20:45:06Z

+        state.set(Cancelled()) >> call.close(status, new Metadata())
+    }
+
+  def unary[F[_]: Async, Request, Response](


Async needed?

ahjohannessen · 2022-02-13T20:45:24Z

+        startCallSync(call, opt)(call => req => call.unary(impl(req, headers), dispatcher)).unsafeRunSync()
+    }
+
+  def stream[F[_]: Async, Request, Response](


Async needed?

Sync is enough.

ahjohannessen · 2022-02-13T22:06:02Z

@naoh87 I have cut a release with your changes. Btw, fs2.grpc.client.ClientSuite.single message to unaryToStreaming is unstable, I think a 50.millis sleep makes it more reliable.

ahjohannessen · 2022-02-16T21:30:08Z

@naoh87 Might be interesting to PR an update to https://github.com/LesnyRumcajs/grpc_bench with latest release :)

naoh87 · 2022-02-18T02:58:25Z

I will do it later.
And fs2-grpc will be gained more performance to use directExecutor safely after all api become non-blocking.

naoh87 commented Feb 5, 2022

View reviewed changes

naoh87 force-pushed the fast_server_runtime branch from 33f708e to c9356cd Compare February 5, 2022 18:14

naoh87 changed the title ~~implement light server runtime~~ Reimplement light server runtime Feb 5, 2022

naoh87 commented Feb 6, 2022

View reviewed changes

naoh87 force-pushed the fast_server_runtime branch from f3be05f to 393eca6 Compare February 6, 2022 10:23

ahjohannessen reviewed Feb 6, 2022

View reviewed changes

naoh87 force-pushed the fast_server_runtime branch from d494558 to 0d0ca45 Compare February 6, 2022 13:50

ahjohannessen reviewed Feb 6, 2022

View reviewed changes

ahjohannessen reviewed Feb 8, 2022

View reviewed changes

ahjohannessen reviewed Feb 10, 2022

View reviewed changes

naoh87 force-pushed the fast_server_runtime branch from 6dd15b8 to fe1fe39 Compare February 11, 2022 05:15

ahjohannessen reviewed Feb 13, 2022

View reviewed changes

naoh87 force-pushed the fast_server_runtime branch from 72cf19c to 78462a1 Compare February 13, 2022 20:27

ahjohannessen reviewed Feb 13, 2022

View reviewed changes

naoh87 force-pushed the fast_server_runtime branch 2 times, most recently from 920a3a3 to fecc2ae Compare February 13, 2022 21:46

implement non-blocking unary request server runtime

c6b4894

naoh87 force-pushed the fast_server_runtime branch from fecc2ae to c6b4894 Compare February 13, 2022 21:47

ahjohannessen merged commit 867782d into typelevel:main Feb 13, 2022

ahjohannessen mentioned this pull request Sep 19, 2022

Remove unbounded queue on server-side stream reception #553

Merged


		import Fs2ServerCall.Cancel

		def stream[F[_]](response: Stream[F, Response], dispatcher: Dispatcher[F])(implicit F: Async[F]): SyncIO[Cancel] =

	def unary[F[_]](response: F[Response], dispatcher: Dispatcher[F])(implicit F: Async[F]): SyncIO[Cancel] =
	def unary[F[_]](response: F[Response], dispatcher: Dispatcher[F])(implicit F: Sync[F]): SyncIO[Cancel] =

	F.delay(call.close(status, metadata))
	close(status, metadata).to[F]

	def unary[F[_]: Async, Request, Response](
	def unary[F[_]: Sync, Request, Response](

	def stream[F[_]: Async, Request, Response](
	def stream[F[_]: Sync, Request, Response](

	object Fs2ServerCall {
	private[server] object Fs2ServerCall {

	final class Fs2ServerCall[Request, Response](
	private[server] final class Fs2ServerCall[Request, Response](

	object Fs2UnaryServerCallHandler {
	private[server] object Fs2UnaryServerCallHandler {

Uh oh!

Uh oh!

Conversation

naoh87 commented Feb 5, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

naoh87 Feb 5, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

naoh87 commented Feb 5, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Benchmark

This PR

fs2-grpc 2.4.4

Akka-gRPC

Uh oh!

ahjohannessen commented Feb 5, 2022

Uh oh!

ahjohannessen commented Feb 5, 2022

Uh oh!

ahjohannessen commented Feb 6, 2022

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!

naoh87 Feb 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

naoh87 Feb 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ahjohannessen commented Feb 6, 2022

Uh oh!

naoh87 commented Feb 8, 2022

Uh oh!

ahjohannessen commented Feb 8, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ahjohannessen commented Feb 8, 2022

Uh oh!

ahjohannessen commented Feb 9, 2022

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

naoh87 commented Feb 5, 2022 •

edited

Loading

naoh87 Feb 5, 2022 •

edited

Loading

naoh87 commented Feb 5, 2022 •

edited

Loading

naoh87 Feb 6, 2022 •

edited

Loading

naoh87 Feb 6, 2022 •

edited

Loading

ahjohannessen Feb 10, 2022 •

edited

Loading

naoh87 commented Feb 11, 2022 •

edited

Loading

naoh87 commented Feb 13, 2022 •

edited

Loading