+bench Add pekko grpc bench

.github/workflows/build.yml

@@ -1355,6 +1355,40 @@ jobs:
       env:
         GRPC_IMAGE_NAME: ${{ needs.set-image-name.outputs.name }}
 
+  scala_pekko_bench:
+    runs-on: ubuntu-latest
+    needs:
+      - set-image-name
+      - changed
+    if: fromJSON(needs.changed.outputs.base) || contains(needs.changed.outputs.files, 'scala_pekko_bench/')
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+
+      - name: Build scala_pekko_bench
+        run: ./build.sh scala_pekko_bench
+        env:
+          GRPC_IMAGE_NAME: ${{ needs.set-image-name.outputs.name }}
+
+      - name: Benchmark scala_pekko_bench
+        run: ./bench.sh scala_pekko_bench
+        env:
+          GRPC_BENCHMARK_DURATION: 30s
+          GRPC_IMAGE_NAME: ${{ needs.set-image-name.outputs.name }}
+
+      - if: github.ref == 'refs/heads/master'
+        name: Log in to GitHub Container Registry
+        uses: docker/login-action@v2
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - if: github.ref == 'refs/heads/master'
+        name: If on master push image to GHCR
+        run: docker push $GRPC_IMAGE_NAME:scala_pekko_bench-complex_proto
+        env:
+          GRPC_IMAGE_NAME: ${{ needs.set-image-name.outputs.name }}
 
   scala_fs2_bench:
     runs-on: ubuntu-latest

scala_pekko_bench/Dockerfile

@@ -0,0 +1,36 @@
+FROM hseeberger/scala-sbt:11.0.7_1.3.13_2.11.12 as BUILDER
+
+WORKDIR /app
+
+# Make sure docker can cache a few static setup things
+
+# initialize only sbt
+RUN mkdir -p /app/project
+COPY scala_pekko_bench/project/build.properties /app/project
+RUN sbt exit
+
+# initialize plugins
+COPY scala_pekko_bench/project /app/project
+RUN sbt exit
+
+# initialize full build
+COPY scala_pekko_bench/*.sbt /app
+# to initialize compiler
+RUN mkdir -p /app/src/main/scala/
+RUN touch /app/src/main/scala/Dummy.scala
+RUN sbt compile
+
+# Actually build project
+COPY scala_pekko_bench/src /app/src
+COPY proto/helloworld/helloworld.proto /app/src/main/protobuf/helloworld.proto
+
+RUN sbt assembly
+
+FROM eclipse-temurin:20.0.1_9-jdk-jammy
+
+ENV GC "-XX:+UseParallelGC"
+ENV _JAVA_OPTIONS "${GC} -XX:MinRAMPercentage=70 -XX:MaxRAMPercentage=70"
+
+COPY --from=builder /app/target/scala-3.3.1/pekko-grpc-quickstart-scala-assembly-1.0.jar .
+
+ENTRYPOINT ["java", "-jar", "pekko-grpc-quickstart-scala-assembly-1.0.jar"]

scala_pekko_bench/build.sbt

@@ -0,0 +1,39 @@
+name := "pekko-grpc-quickstart-scala"
+
+version := "1.0"
+
+scalaVersion := "3.3.1"
+
+run / fork := true
+
+val pekkoVersion = "1.0.1"
+val pekkoHttpVersion = "1.0.0"
+
+enablePlugins(PekkoGrpcPlugin)
+
+// to get latest versions
+resolvers += "Sonatype snapshots" at "https://oss.sonatype.org/content/repositories/snapshots"
+
+libraryDependencies ++= Seq(
+  "ch.qos.logback" % "logback-classic" % "1.2.12",
+  "org.apache.pekko" %% "pekko-actor-typed" % pekkoVersion,
+  "org.apache.pekko" %% "pekko-http" % pekkoHttpVersion,
+  "org.apache.pekko" %% "pekko-http-core" % pekkoHttpVersion,
+  "org.apache.pekko" %% "pekko-parsing" % pekkoHttpVersion,
+  "org.apache.pekko" %% "pekko-stream" % pekkoVersion,
+  "org.apache.pekko" %% "pekko-discovery" % pekkoVersion,
+  "org.apache.pekko" %% "pekko-pki" % pekkoVersion,
+  "org.apache.pekko" %% "pekko-slf4j" % pekkoVersion,
+  "org.apache.pekko" %% "pekko-actor-testkit-typed" % pekkoVersion % Test,
+  "org.apache.pekko" %% "pekko-stream-testkit" % pekkoVersion % Test,
+  "org.scalatest" %% "scalatest" % "3.2.15" % Test
+)
+
+// pekko and Google provided proto files seem to differ a bit so we need to choose
+// (doesn't seem to be important)
+assembly / assemblyMergeStrategy := {
+  case PathList(ps@_*) if ps.last endsWith ".proto" => MergeStrategy.first
+  case x =>
+    val oldStrategy = (assembly / assemblyMergeStrategy).value
+    oldStrategy(x)
+}

scala_pekko_bench/project/build.properties

@@ -0,0 +1 @@
+sbt.version=1.9.4

scala_pekko_bench/project/plugins.sbt

@@ -0,0 +1,5 @@
+addSbtPlugin("org.apache.pekko" % "pekko-grpc-sbt-plugin" % "1.0.0")
+
+addSbtPlugin("com.lightbend.sbt" % "sbt-javaagent" % "0.1.6")
+
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.1")

scala_pekko_bench/src/main/resources/application.conf

@@ -0,0 +1,4 @@
+akka.http.server.max-connections = 1500
+akka.http.server.preview.enable-http2 = on
+akka.http.server.http2.min-collect-strict-entity-size = 1
+akka.actor.default-dispatcher.fork-join-executor.parallelism-max = ${GRPC_SERVER_CPUS}

scala_pekko_bench/src/main/scala/com/example/helloworld/GreeterServer.scala

@@ -0,0 +1,52 @@
+package io.grpc.examples.helloworld
+
+//#import
+
+
+import org.apache.pekko.actor.typed.ActorSystem
+import org.apache.pekko.actor.typed.scaladsl.Behaviors
+import org.apache.pekko.actor.typed.scaladsl.adapter.*
+import org.apache.pekko.http.scaladsl.{Http, HttpConnectionContext}
+import org.apache.pekko.http.scaladsl.model.{HttpRequest, HttpResponse}
+import org.apache.pekko.stream.SystemMaterializer
+
+import scala.concurrent.{ExecutionContext, Future}
+//#import
+
+
+//#server
+object GreeterServer {
+  def main(args: Array[String]): Unit = {
+    val system = ActorSystem[Nothing](Behaviors.empty, "GreeterServer")
+    new GreeterServer()(system).run()
+  }
+}
+
+class GreeterServer(implicit system: ActorSystem[_]) {
+
+  def run(): Future[Http.ServerBinding] = {
+    implicit val ec: ExecutionContext = system.executionContext
+
+    val service: HttpRequest => Future[HttpResponse] =
+      GreeterHandler(new GreeterServiceImpl(system))
+
+    println(s"Parallel: ${system.settings.config.getString("org.apache.pekko.actor.default-dispatcher.fork-join-executor.parallelism-max")} GRPC_SERVER_CPUS: ${sys.env.get("GRPC_SERVER_CPUS")}")
+
+    // org.apache.pekko HTTP 10.1 requires adapters to accept the new actors APIs
+    val bound = Http()(system.toClassic).bindAndHandleAsync(
+      service,
+      interface = "0.0.0.0",
+      port = 50051,
+      connectionContext = HttpConnectionContext()
+    )(SystemMaterializer(system).materializer)
+
+    bound.foreach { binding =>
+      println(s"gRPC server bound to: ${binding.localAddress}")
+    }
+
+    bound
+  }
+  //#server
+
+}
+//#server

scala_pekko_bench/src/main/scala/com/example/helloworld/GreeterServiceImpl.scala

@@ -0,0 +1,34 @@
+package io.grpc.examples.helloworld
+
+//#import
+import scala.concurrent.Future
+
+import org.apache.pekko.NotUsed
+import org.apache.pekko.actor.typed.ActorSystem
+import org.apache.pekko.stream.scaladsl.BroadcastHub
+import org.apache.pekko.stream.scaladsl.Keep
+import org.apache.pekko.stream.scaladsl.MergeHub
+import org.apache.pekko.stream.scaladsl.Sink
+import org.apache.pekko.stream.scaladsl.Source
+
+//#import
+
+//#service-request-reply
+//#service-stream
+class GreeterServiceImpl(system: ActorSystem[_]) extends Greeter {
+  private implicit val sys: ActorSystem[_] = system
+
+  //#service-request-reply
+  val (inboundHub: Sink[HelloRequest, NotUsed], outboundHub: Source[HelloReply, NotUsed]) =
+    MergeHub.source[HelloRequest]
+    .map(request => HelloReply(request.request))
+      .toMat(BroadcastHub.sink[HelloReply])(Keep.both)
+      .run()
+  //#service-request-reply
+
+  override def sayHello(request: HelloRequest): Future[HelloReply] = {
+    Future.successful(HelloReply(request.request))
+  }
+}
+//#service-stream
+//#service-request-reply