Optimize gRPC request processing: fast negotiation path and reduced per-request allocations

codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala

@@ -66,6 +66,8 @@ case class Method(
     else if (ReservedMethodNames.contains(name)) s"$name$ReservedMethodNameSuffix"
     else name
 
+  val handlerName: String = s"${name}Handler"
+
 }
 
 object Method {

codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt

@@ -107,13 +107,19 @@ object @{serviceName}Handler {
           pekko.grpc.scaladsl.ServerReflection.partial(List(@{service.name})))
 }
 
-    private def methodName(request: model.HttpRequest, prefix: String): String =
+    private def methodName(request: model.HttpRequest, prefix: String): String = {
+      // Use Uri.Path structure matching instead of toString to avoid
+      // renderPath + StringBuilder + String allocations per request.
+      // gRPC path format: /{service}/{method}
       request.uri.path match {
-        case model.Uri.Path.Slash(model.Uri.Path.Segment(`prefix`, model.Uri.Path.Slash(model.Uri.Path.Segment(method, model.Uri.Path.Empty)))) =>
-          method
-        case _ =>
-          null
+        case model.Uri.Path.Slash(
+          model.Uri.Path.Segment(serviceName,
+            model.Uri.Path.Slash(
+              model.Uri.Path.Segment(method, model.Uri.Path.Empty))))
+          if serviceName == prefix => method
+        case _ => null
       }
+    }
 
     private def handler(implementation: @serviceName, prefix: String, eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(implicit system: ClassicActorSystemProvider): model.HttpRequest => scala.concurrent.Future[model.HttpResponse] = {
       implicit val mat: Materializer = SystemMaterializer(system).materializer
@@ -122,14 +128,24 @@ object @{serviceName}Handler {
 
       import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
 
+      @if(!powerApis) {
+      // Pre-create lambdas for unary methods to avoid per-request allocation
+      @for(method <- service.methods) {
+      @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
+      val @{method.handlerName}: @method.parameterType => scala.concurrent.Future[@method.outputTypeUnboxed] =
+        (e: @method.parameterType) => implementation.@{method.nameSafe}(e)
+      }
+      }
+      }
+
       def handle(request: model.HttpRequest, method: String): scala.concurrent.Future[model.HttpResponse] =
         GrpcMarshalling.negotiated(request, (reader, writer) =>
           method match {
             @for(method <- service.methods) {
             case "@method.grpcName" =>
                 @{if(powerApis) { "val metadata = MetadataBuilder.fromHeaders(request.headers)" } else { "" }}
                 @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
-                GrpcMarshalling.handleUnary(request.entity, (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }}), eHandler)(
+                GrpcMarshalling.handleUnary(request.entity, @{if(powerApis) { s"(e: ${method.parameterType}) => implementation.${method.nameSafe}(e, metadata)" } else method.handlerName}, eHandler)(
                   @{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, @method.serializer.name, mat, reader, writer, system, ec)
                 } else {
                 @{method.unmarshal}(request.entity)(@{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, mat, reader)
@@ -165,14 +181,24 @@ object @{serviceName}Handler {
 
       import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
 
+      @if(!powerApis) {
+      // Pre-create lambdas for unary methods to avoid per-request allocation
+      @for(method <- service.methods) {
+      @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
+      val @{method.handlerName}: @method.parameterType => scala.concurrent.Future[@method.outputTypeUnboxed] =
+        (e: @method.parameterType) => implementation.@{method.nameSafe}(e)
+      }
+      }
+      }
+
       def handle(request: model.HttpRequest, method: String): scala.concurrent.Future[model.HttpResponse] =
         GrpcMarshalling.negotiated(request, (reader, writer) =>
           method match {
             @for(method <- service.methods) {
             case "@method.grpcName" =>
                 @{if(powerApis) { "val metadata = MetadataBuilder.fromHeaders(request.headers)" } else { "" }}
                 @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
-                GrpcMarshalling.handleUnary(request.entity, (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }}), eHandler)(
+                GrpcMarshalling.handleUnary(request.entity, @{if(powerApis) { s"(e: ${method.parameterType}) => implementation.${method.nameSafe}(e, metadata)" } else method.handlerName}, eHandler)(
                   @{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, @method.serializer.name, mat, reader, writer, system, ec)
                 } else {
                 @{method.unmarshal}(request.entity)(@{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, mat, reader)

runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala

@@ -18,7 +18,7 @@ import pekko.NotUsed
 import pekko.annotation.InternalApi
 import pekko.annotation.InternalStableApi
 import pekko.grpc.GrpcProtocol.{ GrpcProtocolReader, GrpcProtocolWriter }
-import pekko.grpc.internal.{ Codec, Codecs, GrpcProtocolNative, GrpcProtocolWeb, GrpcProtocolWebText }
+import pekko.grpc.internal.{ Codec, Codecs, GrpcProtocolNative, GrpcProtocolWeb, GrpcProtocolWebText, Gzip, Identity }
 import pekko.http.javadsl.{ model => jmodel }
 import pekko.http.scaladsl.model.{ ContentType, HttpHeader, HttpResponse, Trailer }
 import pekko.http.scaladsl.model.HttpEntity.ChunkStreamPart
@@ -141,15 +141,85 @@ object GrpcProtocol {
     case _                                       => None
   }
 
+  // Pre-computed negotiation results for common codec combinations
+  private val NativeIdentityIdentity: (Try[GrpcProtocolReader], GrpcProtocolWriter) =
+    (scala.util.Success(GrpcProtocolNative.newReader(Identity)), GrpcProtocolNative.newWriter(Identity))
+
   /**
    * Calculates the gRPC protocol encoding to use for an interaction with a gRPC client.
+   * Optimized with a fast path for native gRPC that does a single header scan.
    *
    * @param request the client request to respond to.
    * @return the protocol reader for the request, and a protocol writer for the response.
    */
-  def negotiate(request: jmodel.HttpRequest): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] =
-    detect(request).map { variant =>
-      (Codecs.detect(request).map(variant.newReader), variant.newWriter(Codecs.negotiate(request)))
+  def negotiate(request: jmodel.HttpRequest): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = {
+    val mediaType = request.entity.getContentType.mediaType
+    val subType = mediaType.subType
+    val isNative = (subType == "grpc+proto") || (subType == "grpc")
+
+    if (isNative) {
+      // Single-pass header scan for native gRPC
+      var requestEncoding: String = null
+      var acceptEncoding: String = null
+      request match {
+        case sReq: pekko.http.scaladsl.model.HttpMessage =>
+          val headers = sReq.headers
+          var i = 0
+          while (i < headers.size) {
+            val h = headers(i)
+            val name = h.lowercaseName
+            if (name == "grpc-encoding") requestEncoding = h.value
+            else if (name == "grpc-accept-encoding") acceptEncoding = h.value
+            i += 1
+          }
+        case _ =>
+          return slowNegotiateOpt(request, subType)
+      }
+      // Determine reader codec (request encoding)
+      val readerCodec: Codec =
+        if (requestEncoding eq null) Identity
+        else if (requestEncoding == "identity") Identity
+        else if (requestEncoding == "gzip") Gzip
+        else return slowNegotiateOpt(request, subType)
+      // Determine writer codec (response encoding)
+      // Respect client's grpc-accept-encoding preference order (matching Codecs.negotiate).
+      // gRPC format: comma-separated, e.g. "gzip,identity" — client order indicates preference.
+      val writerCodec: Codec =
+        if (acceptEncoding eq null) Identity
+        else if (acceptEncoding == "gzip") Gzip
+        else if (acceptEncoding == "identity") Identity
+        else if (acceptEncoding.contains(",")) {
+          // Multiple encodings: iterate client preference order, pick first supported
+          val parts = acceptEncoding.split(",")
+          var codec: Codec = null
+          var i = 0
+          while (i < parts.length && (codec eq null)) {
+            val trimmed = parts(i).trim
+            if (trimmed == "gzip") codec = Gzip
+            else if (trimmed == "identity") codec = Identity
+            i += 1
+          }
+          if (codec eq null) Identity else codec
+        } else return slowNegotiateOpt(request, subType)
+      // Return pre-computed result for common combinations
+      if ((readerCodec eq Identity) && (writerCodec eq Identity)) return Some(NativeIdentityIdentity)
+      return Some((scala.util.Success(GrpcProtocolNative.newReader(readerCodec)),
+        GrpcProtocolNative.newWriter(writerCodec)))
+    }
+
+    // Non-native protocols - slow path
+    slowNegotiateOpt(request, subType)
+  }
+
+  private def slowNegotiateOpt(request: jmodel.HttpRequest, subType: String)
+      : Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = {
+    val variant = subType match {
+      case "grpc-web" | "grpc-web+proto"           => Some(GrpcProtocolWeb)
+      case "grpc-web-text" | "grpc-web-text+proto" => Some(GrpcProtocolWebText)
+      case "grpc" | "grpc+proto"                   => Some(GrpcProtocolNative)
+      case _                                       => None
     }
+    variant.map(v => (Codecs.detect(request).map(v.newReader), v.newWriter(Codecs.negotiate(request))))
+  }
 
 }

runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala

@@ -27,4 +27,12 @@ trait ProtobufSerializer[T] {
 
 private[grpc] trait ProtobufFrameSerializer[T] extends ProtobufSerializer[T] {
   private[grpc] def serializeDataFrame(t: T): ByteString
+
+  /**
+   * Deserialize a protobuf message from a ByteString starting at the given offset.
+   * Avoids ByteString.slice allocation by passing the offset directly to the parser.
+   * Default implementation falls back to slice + deserialize.
+   */
+  private[grpc] def deserialize(data: ByteString, offset: Int, length: Int): T =
+    deserialize(data.slice(offset, offset + length))
 }

runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala

@@ -100,10 +100,19 @@ object AbstractGrpcProtocol {
       })
       .toContentType
 
+  private val SmallMessageThreshold = 4096
+
   def encodeFrameData(data: ByteString, isCompressed: Boolean, isTrailer: Boolean): ByteString = {
-    val header = new Array[Byte](FrameHeaderSize)
-    writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
-    ByteString.fromArrayUnsafe(header, 0, FrameHeaderSize) ++ data
+    if (data.length <= SmallMessageThreshold) {
+      val combined = new Array[Byte](FrameHeaderSize + data.length)
+      writeFrameHeader(combined, 0, data.length, isCompressed, isTrailer)
+      data.copyToArray(combined, FrameHeaderSize)
+      ByteString.fromArrayUnsafe(combined)
+    } else {
+      val header = new Array[Byte](FrameHeaderSize)
+      writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
+      ByteString.fromArrayUnsafe(header, 0, FrameHeaderSize) ++ data
+    }
   }
 
   def writer(

runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala

@@ -23,8 +23,9 @@ import scala.collection.immutable
 import scala.util.{ Failure, Success, Try }
 
 object Codecs {
-  // TODO should this list be made user-extensible?
-  val supportedCodecs = immutable.Seq(Gzip, Identity)
+  // Codecs supported by this server. The selection order is determined by the
+  // client's grpc-accept-encoding header preference, not by this sequence order.
+  val supportedCodecs = immutable.Seq(Identity, Gzip)
   private val supportedByName: Map[String, Codec] = supportedCodecs.map(c => c.name -> c).toMap
 
   private def extractHeaders(request: jm.HttpMessage): Iterable[jm.HttpHeader] = {

runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala

@@ -44,11 +44,32 @@ object GrpcProtocolNative extends AbstractGrpcProtocol("grpc") {
     AbstractGrpcProtocol.writer(this, codec, encodeFrame(codec, _), encodeDataToResponse(codec))
 
   override protected def reader(codec: Codec): GrpcProtocolReader =
-    AbstractGrpcProtocol.reader(codec, decodeFrame)
+    if (codec eq Identity)
+      AbstractGrpcProtocol.reader(codec, decodeFrame).copy(decodeSingleFrame = decodeIdentitySingleFrame)
+    else AbstractGrpcProtocol.reader(codec, decodeFrame)
 
   @inline
   private def decodeFrame(@nowarn("msg=is never used") frameType: Int, data: ByteString) = DataFrame(data)
 
+  private def decodeIdentitySingleFrame(frame: ByteString): ByteString = {
+    if (frame.length < AbstractGrpcProtocol.FrameHeaderSize) throw new MissingParameterException
+
+    val frameType = frame(0)
+    val length = frame.readIntBE(1)
+    val available = frame.length - AbstractGrpcProtocol.FrameHeaderSize
+    if (length > available) throw new MissingParameterException
+    if (length < 0) throw new IllegalStateException(s"Invalid frame length: $length")
+    if (length < available)
+      throw new IllegalStateException("Unexpected trailing data")
+    if ((frameType & 0x80) != 0) throw new IllegalStateException("Cannot read unknown frame")
+
+    if ((frameType & 1) != 0)
+      throw new io.grpc.StatusException(
+        io.grpc.Status.INTERNAL.withDescription(
+          "Compressed-Flag bit is set, but a compression encoding is not specified"))
+    frame.drop(AbstractGrpcProtocol.FrameHeaderSize)
+  }
+
   @inline
   private def encodeFrame(codec: Codec, frame: Frame): ChunkStreamPart =
     frame match {

runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala

@@ -60,7 +60,25 @@ object GrpcMarshalling {
       reader: GrpcProtocolReader): CompletionStage[T] =
     entity match {
       case strict: pekko.http.scaladsl.model.HttpEntity.Strict =>
-        completedOrFailed(u.deserialize(reader.decodeSingleFrame(strict.data)))
+        completedOrFailed {
+          u match {
+            case frameSerializer: ProtobufFrameSerializer[T @unchecked] =>
+              if (strict.data.length < AbstractGrpcProtocol.FrameHeaderSize)
+                throw new MissingParameterException
+              val frameType = strict.data(0)
+              if ((frameType & 1) != 0)
+                throw new io.grpc.StatusException(
+                  io.grpc.Status.INTERNAL.withDescription(
+                    "Compressed-Flag bit is set, but a compression encoding is not specified"))
+              val declaredLength = strict.data.readIntBE(1)
+              val available = strict.data.length - AbstractGrpcProtocol.FrameHeaderSize
+              if (declaredLength > available) throw new MissingParameterException
+              if (declaredLength < 0) throw new IllegalStateException(s"Invalid frame length: $declaredLength")
+              frameSerializer.deserialize(strict.data, AbstractGrpcProtocol.FrameHeaderSize, declaredLength)
+            case _ =>
+              u.deserialize(reader.decodeSingleFrame(strict.data))
+          }
+        }
       case _ =>
         unmarshal(entity.getDataBytes, u, mat, reader)
     }

runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala

@@ -55,9 +55,10 @@ object GrpcMarshalling {
   }
 
   def negotiated[T](req: HttpRequest, f: (GrpcProtocolReader, GrpcProtocolWriter) => Future[T]): Option[Future[T]] =
-    GrpcProtocol.negotiate(req).map {
-      case (Success(reader), writer) => f(reader, writer)
-      case (Failure(ex), _)          => Future.failed(ex)
+    GrpcProtocol.negotiate(req) match {
+      case Some((Success(reader), writer)) => Some(f(reader, writer))
+      case Some((Failure(ex), _))          => Some(Future.failed(ex))
+      case None                            => None
     }
 
   def unmarshal[T](data: Source[ByteString, Any])(
@@ -125,18 +126,35 @@ object GrpcMarshalling {
     entity match {
       case HttpEntity.Strict(_, data) =>
         try {
-          val in = u.deserialize(reader.decodeSingleFrame(data))
+          val in = u match {
+            case frameSerializer: ProtobufFrameSerializer[In @unchecked] =>
+              if (data.length < AbstractGrpcProtocol.FrameHeaderSize)
+                throw new MissingParameterException
+              val frameType = data(0)
+              if ((frameType & 1) != 0)
+                throw new io.grpc.StatusException(
+                  io.grpc.Status.INTERNAL.withDescription(
+                    "Compressed-Flag bit is set, but a compression encoding is not specified"))
+              val declaredLength = data.readIntBE(1)
+              val available = data.length - AbstractGrpcProtocol.FrameHeaderSize
+              if (declaredLength > available) throw new MissingParameterException
+              if (declaredLength < 0) throw new IllegalStateException(s"Invalid frame length: $declaredLength")
+              frameSerializer.deserialize(data, AbstractGrpcProtocol.FrameHeaderSize, declaredLength)
+            case _ =>
+              u.deserialize(reader.decodeSingleFrame(data))
+          }
           invokeUnary(in, implementation, eHandler)
         } catch {
-          case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+          case NonFatal(ex) => handleUnaryException(ex, eHandler)
         }
       case _ =>
         val requestFuture = unmarshal[In](entity)(u, mat, reader)
         requestFuture.value match {
           case Some(Success(in)) => invokeUnary(in, implementation, eHandler)
-          case Some(Failure(ex)) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+          case Some(Failure(ex)) => handleUnaryException(ex, eHandler)
           case None              =>
-            val exceptionHandler = unaryExceptionHandler(eHandler)
+            // Lazy: only create the PartialFunction if actually needed for recovery
+            lazy val exceptionHandler = unaryExceptionHandler(eHandler)
             requestFuture
               .flatMap(in => invokeUnary(in, implementation, eHandler))
               .recoverWith(exceptionHandler)
@@ -154,7 +172,7 @@ object GrpcMarshalling {
       ec: ExecutionContext): Future[HttpResponse] =
     try handleUnaryResponse(implementation(in), eHandler)
     catch {
-      case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case NonFatal(ex) => handleUnaryException(ex, eHandler)
     }
 
   @inline private def handleUnaryResponse[Out](
@@ -166,9 +184,9 @@ object GrpcMarshalling {
       ec: ExecutionContext): Future[HttpResponse] =
     responseFuture.value match {
       case Some(Success(out)) => marshalUnaryResponse(out, eHandler)
-      case Some(Failure(ex))  => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case Some(Failure(ex))  => handleUnaryException(ex, eHandler)
       case None               =>
-        val exceptionHandler = unaryExceptionHandler(eHandler)
+        lazy val exceptionHandler = unaryExceptionHandler(eHandler)
         responseFuture.map(out => marshal[Out](out, eHandler)(m, writer, system)).recoverWith(exceptionHandler)
     }
 
@@ -180,9 +198,25 @@ object GrpcMarshalling {
       system: ClassicActorSystemProvider): Future[HttpResponse] =
     try FastFuture.successful(marshal[Out](out, eHandler)(m, writer, system))
     catch {
-      case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case NonFatal(ex) => handleUnaryException(ex, eHandler)
     }
 
+  /**
+   * Inline exception handling for synchronous error paths.
+   * Avoids allocating a PartialFunction via GrpcExceptionHandler.from() for each exception.
+   * Instead, directly applies the mapper and converts to response.
+   */
+  @inline private def handleUnaryException(
+      ex: Throwable,
+      eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(
+      implicit writer: GrpcProtocolWriter,
+      system: ClassicActorSystemProvider): Future[HttpResponse] = {
+    val mapper = eHandler(system.classicSystem)
+    val trailers = if (mapper.isDefinedAt(ex)) mapper(ex)
+    else GrpcExceptionHandler.defaultMapper(system.classicSystem)(ex)
+    FastFuture.successful(GrpcResponseHelpers.status(trailers))
+  }
+
   @inline private def unaryExceptionHandler(eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(
       implicit system: ClassicActorSystemProvider,
       writer: GrpcProtocolWriter): PartialFunction[Throwable, Future[HttpResponse]] =