This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch optimize/adaptive-compression in repository https://gitbox.apache.org/repos/asf/pekko-grpc.git
commit 6c18e672b1b2709174b8a7d0a9895398d8a54d97 Author: 虎鸣 <[email protected]> AuthorDate: Wed Jun 24 00:04:19 2026 +0800 feat: add adaptive gzip compression for gRPC responses Motivation: The previous PR always used Identity (no compression) for native gRPC responses to avoid CPU overhead on small messages. However, large messages benefit significantly from gzip compression. An adaptive approach that compresses only messages above a configurable threshold provides the best of both worlds. Modification: - Add AdaptiveGzip codec that only compresses messages above a configurable threshold (default 1024 bytes). Messages below the threshold pass through uncompressed while the grpc-encoding header still advertises gzip. - Add Codec.compressWithFlag() returning (ByteString, Boolean) to signal per-frame compression status accurately in the gRPC frame header (bit 0 of the 5-byte header). This fixes a correctness bug where streaming frames would incorrectly set the compression flag even for uncompressed data. - Add pekko.grpc.server.compression-threshold config (default 1024). Generated Scala handlers read config once at handler creation and pass the threshold as Int to avoid per-request config overhead. - Pre-compute NativeIdentityAdaptiveGzip negotiation result and cache the default AdaptiveGzip instance for zero per-request allocation on the common path. - Add ProtobufFrameSerializer.serializedDataSize() for efficient size checking without double serialization. - Add adaptive fast path in GrpcResponseHelpers for small messages using serializeDataFrame single-allocation path. Result: - Messages below threshold: zero compression CPU overhead, sent uncompressed with correct frame header flag. - Messages above threshold: gzip compressed with correct frame header flag. Per gRPC spec, per-frame compression flag tells clients whether each frame is compressed. - Default threshold of 1024 bytes aligns with industry consensus (Netty, Ktor, OneUptime gRPC guide) as the break-even point where gzip CPU cost is justified by bandwidth savings on protobuf data. - JIT/GC optimized: zero per-request allocation on common path via pre-computed negotiation result and cached AdaptiveGzip singleton. Tests: - runtime / Test / test - 152 tests passed - plugin-tester-scala / Test / testOnly ErrorReportingSpec - passed - 30 new AdaptiveGzip tests covering codec behavior, boundary conditions, compressWithFlag, streaming frame header correctness, round-trip encode/decode, instance caching, and negotiate integration References: Refs #739 --- .../twirl/templates/ScalaServer/Handler.scala.txt | 6 +- runtime/src/main/resources/reference.conf | 111 +++++---- .../scala/org/apache/pekko/grpc/GrpcProtocol.scala | 64 ++++- .../org/apache/pekko/grpc/ProtobufSerializer.scala | 10 + .../apache/pekko/grpc/internal/AdaptiveGzip.scala | 80 ++++++ .../org/apache/pekko/grpc/internal/Codec.scala | 10 + .../pekko/grpc/internal/GrpcProtocolNative.scala | 9 +- .../pekko/grpc/internal/GrpcResponseHelpers.scala | 21 +- .../grpc/javadsl/GoogleProtobufSerializer.scala | 1 + .../pekko/grpc/scaladsl/GrpcMarshalling.scala | 15 ++ .../grpc/scaladsl/ScalapbProtobufSerializer.scala | 1 + .../pekko/grpc/internal/AdaptiveGzipSpec.scala | 270 +++++++++++++++++++++ 12 files changed, 528 insertions(+), 70 deletions(-) diff --git a/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt b/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt index 196413e4..61df22b9 100644 --- a/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt +++ b/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt @@ -125,6 +125,7 @@ object @{serviceName}Handler { implicit val mat: Materializer = SystemMaterializer(system).materializer implicit val ec: ExecutionContext = mat.executionContext val spi = TelemetryExtension(system).spi + val compressionThreshold = system.classicSystem.settings.config.getInt("pekko.grpc.server.compression-threshold") import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport} @@ -156,7 +157,7 @@ object @{serviceName}Handler { } case m => GrpcExceptionHandler.from(eHandler(system.classicSystem))(system, writer)(new NotImplementedError(s"Not implemented: $m")) } - ).getOrElse(unsupportedMediaType) + , compressionThreshold).getOrElse(unsupportedMediaType) request => { val method = methodName(request, prefix) @@ -178,6 +179,7 @@ object @{serviceName}Handler { implicit val mat: Materializer = SystemMaterializer(system).materializer implicit val ec: ExecutionContext = mat.executionContext val spi = TelemetryExtension(system).spi + val compressionThreshold = system.classicSystem.settings.config.getInt("pekko.grpc.server.compression-threshold") import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport} @@ -209,7 +211,7 @@ object @{serviceName}Handler { } case m => GrpcExceptionHandler.from(eHandler(system.classicSystem))(system, writer)(new NotImplementedError(s"Not implemented: $m")) } - ).getOrElse(unsupportedMediaType) + , compressionThreshold).getOrElse(unsupportedMediaType) Function.unlift((req: model.HttpRequest) => { val method = methodName(req, prefix) diff --git a/runtime/src/main/resources/reference.conf b/runtime/src/main/resources/reference.conf index 99fcc62a..9085cfc2 100644 --- a/runtime/src/main/resources/reference.conf +++ b/runtime/src/main/resources/reference.conf @@ -1,56 +1,69 @@ # SPDX-License-Identifier: Apache-2.0 //#defaults -pekko.grpc.client."*" { - # netty or pekko-http (experimental) - backend = "netty" - - # Host to use if service-discovery-mechanism is set to static or grpc-dns - host = "" - - service-discovery { - mechanism = "static" - # Service name to use if a service-discovery.mechanism other than static or grpc-dns - service-name = "" - # See https://pekko.apache.org/docs/pekko-management/current/discovery/index.html for meanings for each mechanism - # if blank then not passed to the lookup - port-name = "" - protocol = "" - - # timeout for service discovery resolving - resolve-timeout = 1s +pekko.grpc { + # Server-side configuration for gRPC services + server { + # Adaptive compression: messages smaller than this threshold (in bytes) + # are sent uncompressed, while larger messages are compressed with gzip. + # This avoids the CPU overhead of gzip for small messages where the + # bandwidth savings are negligible. + # Set to 0 to always compress (every message goes through gzip). + # Set to a very large value (e.g. 2147483647) to effectively disable compression. + compression-threshold = 1024 } - # port to use if service-discovery-mechanism is static or service discovery does not return a port - port = 0 - - # Experimental in grpc-java https://github.com/grpc/grpc-java/issues/1771 - # pick_first or round_robin - load-balancing-policy = "" - - deadline = infinite - override-authority = "" - user-agent = "" - # Location on the classpath of CA PEM to trust - trusted = "" - use-tls = true - # SSL provider to use: - # leave empty to auto-detect, or configure 'jdk' or 'openssl'. - ssl-provider = "" - - # TODO: Enforce HTTP/2 TLS restrictions: https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-9.2 - - # The number of times to try connecting before giving up. - # '-1': means retry indefinitely, '0' is invalid, '1' means fail - # after the first failed attempt. - # When load balancing we don't count individual connection - # failures, so in that case any value larger than '1' is also - # interpreted as retrying 'indefinitely'. - connection-attempts = 20 - - # Service discovery mechanism to use. The default is to use a static host - # and port that will be resolved via DNS. - # Any of the mechanisms described in https://pekko.apache.org/docs/pekko-management/current/discovery/index.html can be used - # including Kubernetes, Consul, AWS API + client."*" { + # netty or pekko-http (experimental) + backend = "netty" + + # Host to use if service-discovery-mechanism is set to static or grpc-dns + host = "" + + service-discovery { + mechanism = "static" + # Service name to use if a service-discovery.mechanism other than static or grpc-dns + service-name = "" + # See https://pekko.apache.org/docs/pekko-management/current/discovery/index.html for meanings for each mechanism + # if blank then not passed to the lookup + port-name = "" + protocol = "" + + # timeout for service discovery resolving + resolve-timeout = 1s + } + + # port to use if service-discovery-mechanism is static or service discovery does not return a port + port = 0 + + # Experimental in grpc-java https://github.com/grpc/grpc-java/issues/1771 + # pick_first or round_robin + load-balancing-policy = "" + + deadline = infinite + override-authority = "" + user-agent = "" + # Location on the classpath of CA PEM to trust + trusted = "" + use-tls = true + # SSL provider to use: + # leave empty to auto-detect, or configure 'jdk' or 'openssl'. + ssl-provider = "" + + # TODO: Enforce HTTP/2 TLS restrictions: https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-9.2 + + # The number of times to try connecting before giving up. + # '-1': means retry indefinitely, '0' is invalid, '1' means fail + # after the first failed attempt. + # When load balancing we don't count individual connection + # failures, so in that case any value larger than '1' is also + # interpreted as retrying 'indefinitely'. + connection-attempts = 20 + + # Service discovery mechanism to use. The default is to use a static host + # and port that will be resolved via DNS. + # Any of the mechanisms described in https://pekko.apache.org/docs/pekko-management/current/discovery/index.html can be used + # including Kubernetes, Consul, AWS API + } } //#defaults diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala index c0372e9c..4bf7da08 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala +++ b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala @@ -18,7 +18,16 @@ 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, Gzip, Identity } +import pekko.grpc.internal.{ + AdaptiveGzip, + 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,18 +150,43 @@ object GrpcProtocol { case _ => None } - // Pre-computed negotiation results for common codec combinations + // Pre-computed negotiation result for Identity reader + Identity writer (no gzip support) private val NativeIdentityIdentity: (Try[GrpcProtocolReader], GrpcProtocolWriter) = (scala.util.Success(GrpcProtocolNative.newReader(Identity)), GrpcProtocolNative.newWriter(Identity)) + // Pre-computed negotiation result for Identity reader + AdaptiveGzip writer (default threshold) + // This is the most common case: client accepts gzip, server uses adaptive compression. + private val NativeIdentityAdaptiveGzip: (Try[GrpcProtocolReader], GrpcProtocolWriter) = + (scala.util.Success(GrpcProtocolNative.newReader(Identity)), + GrpcProtocolNative.newWriter(AdaptiveGzip(AdaptiveGzip.DefaultCompressionThreshold))) + + /** + * 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. + * + * For response encoding, uses adaptive gzip: messages above a configurable threshold + * (default 1KB) are compressed with gzip, while smaller messages are sent uncompressed. + * The gRPC per-frame compression flag (bit 0 of the 5-byte frame header) tells the + * client whether each individual frame is compressed. + * + * @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)] = + negotiate(request, AdaptiveGzip.DefaultCompressionThreshold) + /** * 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. + * @param compressionThreshold minimum message size in bytes to trigger gzip compression. + * Messages below this threshold are sent uncompressed. * @return the protocol reader for the request, and a protocol writer for the response. */ - def negotiate(request: jmodel.HttpRequest): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = { + def negotiate( + request: jmodel.HttpRequest, + compressionThreshold: Int): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = { val mediaType = request.entity.getContentType.mediaType val subType = mediaType.subType val isNative = (subType == "grpc+proto") || (subType == "grpc") @@ -160,6 +194,7 @@ object GrpcProtocol { 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 @@ -168,6 +203,7 @@ object GrpcProtocol { 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 _ => @@ -179,15 +215,19 @@ object GrpcProtocol { else if (requestEncoding == "gzip") Gzip else return slowNegotiateOpt(request, subType) // Determine writer codec (response encoding) - // For native gRPC, always prefer Identity for responses. The per-frame compression - // flag tells the client whether each frame is compressed. For typical small gRPC - // messages, avoiding compression saves significant CPU overhead (~20-30% of handler time). - // Clients that need compression for large messages can still handle uncompressed frames. - val writerCodec: Codec = Identity - // 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))) + // Adaptive compression: use gzip for messages above threshold, identity for smaller ones. + // Only activate when client advertises gzip support via grpc-accept-encoding. + val clientAcceptsGzip = acceptEncoding != null && acceptEncoding.contains("gzip") + if (!clientAcceptsGzip && (readerCodec eq Identity)) return Some(NativeIdentityIdentity) + if (clientAcceptsGzip && (readerCodec eq Identity) && + compressionThreshold == AdaptiveGzip.DefaultCompressionThreshold) + return Some(NativeIdentityAdaptiveGzip) + val writerCodec: Codec = + if (clientAcceptsGzip) AdaptiveGzip(compressionThreshold) + else Identity + return Some( + (scala.util.Success(GrpcProtocolNative.newReader(readerCodec)), + GrpcProtocolNative.newWriter(writerCodec))) } // Non-native protocols - slow path diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala b/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala index cbc2acc0..14803283 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala +++ b/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala @@ -35,4 +35,14 @@ private[grpc] trait ProtobufFrameSerializer[T] extends ProtobufSerializer[T] { */ private[grpc] def deserialize(data: ByteString, offset: Int, length: Int): T = deserialize(data.slice(offset, offset + length)) + + /** + * Returns the serialized size of the message without actually serializing it. + * Used by adaptive compression to decide whether to compress small messages. + * + * Implementations should override this with an efficient method (e.g. + * `GeneratedMessage.serializedSize` for ScalaPB or `Message.getSerializedSize` + * for Google protobuf) to avoid the overhead of full serialization. + */ + private[grpc] def serializedDataSize(t: T): Int = serialize(t).length } diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AdaptiveGzip.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AdaptiveGzip.scala new file mode 100644 index 00000000..48e3b5c3 --- /dev/null +++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AdaptiveGzip.scala @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.pekko.grpc.internal + +import java.io.ByteArrayOutputStream +import java.util.zip.GZIPOutputStream + +import org.apache.pekko.annotation.InternalApi +import org.apache.pekko.util.ByteString + +/** + * Adaptive gzip codec that only compresses messages above a size threshold. + * + * For small messages, the CPU overhead of gzip compression exceeds the bandwidth + * savings. This codec checks the message size before compressing and passes through + * uncompressed data for small messages, while still advertising "gzip" as the + * encoding to the client. The gRPC per-frame compression flag (bit 0 of the + * 5-byte frame header) tells the client whether each individual frame is compressed. + * + * @param compressionThreshold minimum message size in bytes to trigger compression + */ +@InternalApi +private[grpc] class AdaptiveGzip(val compressionThreshold: Int) extends Codec { + override val name: String = "gzip" + + override def compress(uncompressed: ByteString): ByteString = { + if (uncompressed.size < compressionThreshold) uncompressed + else { + val baos = new ByteArrayOutputStream(uncompressed.size) + val gzos = new GZIPOutputStream(baos) + try gzos.write(uncompressed.toArrayUnsafe()) + finally gzos.close() + ByteString.fromArrayUnsafe(baos.toByteArray) + } + } + + override def uncompress(compressed: ByteString): ByteString = Gzip.uncompress(compressed) + + override def uncompress(compressedBitSet: Boolean, bytes: ByteString): ByteString = + Gzip.uncompress(compressedBitSet, bytes) + + override def isCompressed: Boolean = true + + override def compressWithFlag(bytes: ByteString): (ByteString, Boolean) = { + if (bytes.size < compressionThreshold) (bytes, false) + else { + val baos = new ByteArrayOutputStream(bytes.size) + val gzos = new GZIPOutputStream(baos) + try gzos.write(bytes.toArrayUnsafe()) + finally gzos.close() + (ByteString.fromArrayUnsafe(baos.toByteArray), true) + } + } +} + +@InternalApi +private[grpc] object AdaptiveGzip { + val DefaultCompressionThreshold: Int = 1024 + + private val DefaultInstance: AdaptiveGzip = new AdaptiveGzip(DefaultCompressionThreshold) + + def apply(threshold: Int = DefaultCompressionThreshold): AdaptiveGzip = + if (threshold == DefaultCompressionThreshold) DefaultInstance + else new AdaptiveGzip(threshold) +} diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala index 24b31a8f..d4953947 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala +++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala @@ -28,4 +28,14 @@ abstract class Codec { def uncompress(compressedBitSet: Boolean, bytes: ByteString): ByteString def isCompressed: Boolean = this != Identity + + /** + * Compress data and report whether compression was actually applied. + * Returns a tuple of (compressedData, wasCompressed) where wasCompressed + * indicates whether the data should be marked with the compression flag + * in the gRPC frame header. Default implementation uses compress() + isCompressed. + * Adaptive codecs override this to signal per-frame compression decisions. + */ + def compressWithFlag(bytes: ByteString): (ByteString, Boolean) = + (compress(bytes), isCompressed) } diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala index a2263c5c..031a8a79 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala +++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala @@ -69,7 +69,8 @@ object GrpcProtocolNative extends AbstractGrpcProtocol("grpc") { private def encodeFrame(codec: Codec, frame: Frame): ChunkStreamPart = frame match { case DataFrame(data) => - Chunk(AbstractGrpcProtocol.encodeFrameData(codec.compress(data), codec.isCompressed, isTrailer = false)) + val (compressed, wasCompressed) = codec.compressWithFlag(data) + Chunk(AbstractGrpcProtocol.encodeFrameData(compressed, wasCompressed, isTrailer = false)) case TrailerFrame(headers) => LastChunk(trailer = headers) } private def encodeDataToResponse( @@ -81,6 +82,8 @@ object GrpcProtocolNative extends AbstractGrpcProtocol("grpc") { protocol = HttpProtocols.`HTTP/1.1`, attributes = Map.empty[AttributeKey[?], Any].updated(AttributeKeys.trailer, trailer)) - private def encodeDataToFrameBytes(codec: Codec, data: ByteString): ByteString = - AbstractGrpcProtocol.encodeFrameData(codec.compress(data), codec.isCompressed, isTrailer = false) + private def encodeDataToFrameBytes(codec: Codec, data: ByteString): ByteString = { + val (compressed, wasCompressed) = codec.compressWithFlag(data) + AbstractGrpcProtocol.encodeFrameData(compressed, wasCompressed, isTrailer = false) + } } diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala index 3e285b73..93de4052 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala +++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala @@ -72,10 +72,23 @@ object GrpcResponseHelpers { system: ClassicActorSystemProvider): HttpResponse = { val responseHeaders = responseHeadersFor(writer) try { - if ((writer.messageEncoding eq Identity) && writer.contentType == GrpcProtocolNative.contentType) { - m match { - case frameSerializer: ProtobufFrameSerializer[T @unchecked] => - nativeResponse(writer, frameSerializer.serializeDataFrame(e), responseHeaders) + if (writer.contentType == GrpcProtocolNative.contentType) { + writer.messageEncoding match { + case Identity => + m match { + case frameSerializer: ProtobufFrameSerializer[T @unchecked] => + nativeResponse(writer, frameSerializer.serializeDataFrame(e), responseHeaders) + case _ => + writer.encodeDataToResponse(m.serialize(e), responseHeaders, TrailerOkAttribute) + } + case adaptive: AdaptiveGzip => + m match { + case frameSerializer: ProtobufFrameSerializer[T @unchecked] + if frameSerializer.serializedDataSize(e) < adaptive.compressionThreshold => + nativeResponse(writer, frameSerializer.serializeDataFrame(e), responseHeaders) + case _ => + writer.encodeDataToResponse(m.serialize(e), responseHeaders, TrailerOkAttribute) + } case _ => writer.encodeDataToResponse(m.serialize(e), responseHeaders, TrailerOkAttribute) } diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala index c71b6ed1..ec35b123 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala +++ b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala @@ -28,6 +28,7 @@ class GoogleProtobufSerializer[T <: com.google.protobuf.Message](parser: Parser[ override def serialize(t: T): ByteString = ByteString.fromArrayUnsafe(t.toByteArray) + override private[grpc] def serializedDataSize(t: T): Int = t.getSerializedSize override private[grpc] def serializeDataFrame(t: T): ByteString = { val dataLength = t.getSerializedSize val frame = new Array[Byte](AbstractGrpcProtocol.FrameHeaderSize + dataLength) diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala index 6d5937fa..65042f12 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala +++ b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala @@ -61,6 +61,21 @@ object GrpcMarshalling { case None => None } + /** + * Like `negotiated` but accepts a pre-read compression threshold to avoid + * per-request config reading. The threshold should be read once at handler + * creation time from `pekko.grpc.server.compression-threshold`. + */ + def negotiated[T]( + req: HttpRequest, + f: (GrpcProtocolReader, GrpcProtocolWriter) => Future[T], + compressionThreshold: Int): Option[Future[T]] = + GrpcProtocol.negotiate(req, compressionThreshold) 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])( implicit u: ProtobufSerializer[T], mat: Materializer, diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala index 99450eb5..8aa589c0 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala +++ b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala @@ -29,6 +29,7 @@ class ScalapbProtobufSerializer[T <: GeneratedMessage](companion: GeneratedMessa extends ProtobufFrameSerializer[T] { override def serialize(t: T): ByteString = ByteString.fromArrayUnsafe(t.toByteArray) + override private[grpc] def serializedDataSize(t: T): Int = t.serializedSize override private[grpc] def serializeDataFrame(t: T): ByteString = { val dataLength = t.serializedSize val frame = new Array[Byte](AbstractGrpcProtocol.FrameHeaderSize + dataLength) diff --git a/runtime/src/test/scala/org/apache/pekko/grpc/internal/AdaptiveGzipSpec.scala b/runtime/src/test/scala/org/apache/pekko/grpc/internal/AdaptiveGzipSpec.scala new file mode 100644 index 00000000..8e7ed558 --- /dev/null +++ b/runtime/src/test/scala/org/apache/pekko/grpc/internal/AdaptiveGzipSpec.scala @@ -0,0 +1,270 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package org.apache.pekko.grpc.internal + +import org.apache.pekko +import pekko.grpc.GrpcProtocol +import pekko.grpc.scaladsl.headers +import pekko.http.scaladsl.model.{ ContentTypes, HttpHeader, HttpRequest } +import pekko.util.ByteString +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import scala.collection.immutable + +class AdaptiveGzipSpec extends AnyWordSpec with Matchers { + + private val smallData = ByteString.fromArrayUnsafe(new Array[Byte](100)) + private val largeData = ByteString.fromArrayUnsafe(new Array[Byte](4096)) + + private def nativeRequest(acceptEncoding: String = null, encoding: String = null): HttpRequest = { + var hdrs: immutable.Seq[HttpHeader] = immutable.Seq.empty + if (acceptEncoding != null) + hdrs = hdrs :+ headers.`Message-Accept-Encoding`(acceptEncoding) + if (encoding != null) + hdrs = hdrs :+ headers.`Message-Encoding`(encoding) + HttpRequest( + headers = hdrs, + entity = pekko.http.scaladsl.model.HttpEntity(ContentTypes.`application/grpc+proto`, ByteString.empty)) + } + + "AdaptiveGzip" should { + + "not compress data below threshold" in { + val codec = AdaptiveGzip(1024) + val result = codec.compress(smallData) + result should be theSameInstanceAs smallData + } + + "compress data above threshold" in { + val codec = AdaptiveGzip(1024) + val result = codec.compress(largeData) + result should not be theSameInstanceAs(largeData) + result.length should be < largeData.length + } + + "use default threshold of 1024" in { + val codec = AdaptiveGzip() + codec.compressionThreshold should be(1024) + } + + "report name as gzip" in { + AdaptiveGzip().name should be("gzip") + } + + "report isCompressed as true" in { + AdaptiveGzip().isCompressed should be(true) + } + + "uncompress data compressed by Gzip" in { + val codec = AdaptiveGzip(1024) + val compressed = Gzip.compress(largeData) + val uncompressed = codec.uncompress(compressed) + uncompressed should be(largeData) + } + + "uncompress with compressedBitSet=true" in { + val codec = AdaptiveGzip(1024) + val compressed = Gzip.compress(largeData) + val uncompressed = codec.uncompress(compressedBitSet = true, compressed) + uncompressed should be(largeData) + } + + "pass through with compressedBitSet=false" in { + val codec = AdaptiveGzip(1024) + val uncompressed = codec.uncompress(compressedBitSet = false, smallData) + uncompressed should be theSameInstanceAs smallData + } + + "round-trip: compress then uncompress preserves data" in { + val codec = AdaptiveGzip(1024) + val data = ByteString("Hello, World! " * 200) + val compressed = codec.compress(data) + val uncompressed = codec.uncompress(compressedBitSet = true, compressed) + uncompressed should be(data) + } + + "round-trip: pass-through for small data" in { + val codec = AdaptiveGzip(1024) + val data = ByteString("small") + val result = codec.compress(data) + result should be theSameInstanceAs data + codec.uncompress(compressedBitSet = false, result) should be(data) + } + + "not compress empty data" in { + val codec = AdaptiveGzip(1024) + val data = ByteString.empty + val result = codec.compress(data) + result should be theSameInstanceAs data + } + + "always compress when threshold is 0" in { + val codec = AdaptiveGzip(0) + val data = ByteString("tiny") + val result = codec.compress(data) + result should not be theSameInstanceAs(data) + codec.uncompress(compressedBitSet = true, result) should be(data) + } + + "never compress when threshold is Int.MaxValue" in { + val codec = AdaptiveGzip(Int.MaxValue) + val result = codec.compress(largeData) + result should be theSameInstanceAs largeData + } + + "compress data at exactly the threshold (boundary)" in { + val codec = AdaptiveGzip(1024) + val data = ByteString.fromArrayUnsafe(new Array[Byte](1024)) + val result = codec.compress(data) + result should not be theSameInstanceAs(data) + } + + "not compress data at threshold minus 1 (boundary)" in { + val codec = AdaptiveGzip(1024) + val data = ByteString.fromArrayUnsafe(new Array[Byte](1023)) + val result = codec.compress(data) + result should be theSameInstanceAs data + } + + "compressWithFlag returns false for small data" in { + val codec = AdaptiveGzip(1024) + val (data, flag) = codec.compressWithFlag(smallData) + data should be theSameInstanceAs smallData + flag should be(false) + } + + "compressWithFlag returns true for large data" in { + val codec = AdaptiveGzip(1024) + val (data, flag) = codec.compressWithFlag(largeData) + data should not be theSameInstanceAs(largeData) + flag should be(true) + } + + "cache default instance" in { + val a = AdaptiveGzip() + val b = AdaptiveGzip() + a should be theSameInstanceAs b + } + + "not cache non-default instances" in { + val a = AdaptiveGzip(2048) + val b = AdaptiveGzip(2048) + a should not be theSameInstanceAs(b) + } + } + + "GrpcProtocol.negotiate" should { + + "use AdaptiveGzip when client accepts gzip" in { + val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip")) + result shouldBe defined + val (_, writer) = result.get + writer.messageEncoding shouldBe an[AdaptiveGzip] + writer.messageEncoding.name should be("gzip") + } + + "use AdaptiveGzip when client accepts gzip among multiple encodings" in { + val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "identity,gzip")) + result shouldBe defined + val (_, writer) = result.get + writer.messageEncoding shouldBe an[AdaptiveGzip] + } + + "use Identity when client does not accept gzip" in { + val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "identity")) + result shouldBe defined + val (_, writer) = result.get + writer.messageEncoding should be(Identity) + } + + "use Identity when no accept-encoding header" in { + val result = GrpcProtocol.negotiate(nativeRequest()) + result shouldBe defined + val (_, writer) = result.get + writer.messageEncoding should be(Identity) + } + + "use Gzip reader when request has gzip encoding" in { + val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip", encoding = "gzip")) + result shouldBe defined + val (readerTry, writer) = result.get + val reader = readerTry.get + reader.messageEncoding should be(Gzip) + writer.messageEncoding shouldBe an[AdaptiveGzip] + } + + "use Identity reader when request has no encoding" in { + val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip")) + result shouldBe defined + val (readerTry, _) = result.get + val reader = readerTry.get + reader.messageEncoding should be(Identity) + } + + "use custom compression threshold when specified" in { + val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip"), 2048) + result shouldBe defined + val (_, writer) = result.get + writer.messageEncoding shouldBe an[AdaptiveGzip] + writer.messageEncoding.asInstanceOf[AdaptiveGzip].compressionThreshold should be(2048) + } + + "use default threshold of 1024 when no threshold specified" in { + val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip")) + result shouldBe defined + val (_, writer) = result.get + writer.messageEncoding.asInstanceOf[AdaptiveGzip].compressionThreshold should be(1024) + } + } + + "Streaming frame encoding with AdaptiveGzip" should { + + "set compression flag to 0 for small messages (below threshold)" in { + val codec = AdaptiveGzip(1024) + val smallMsg = ByteString.fromArrayUnsafe(new Array[Byte](100)) + val writer = GrpcProtocolNative.newWriter(codec) + val frame = writer.encodeFrame(GrpcProtocol.DataFrame(smallMsg)) + val chunk = frame.asInstanceOf[pekko.http.scaladsl.model.HttpEntity.Chunk] + val frameData = chunk.data + frameData(0) should be(0.toByte) + frameData.length should be(AbstractGrpcProtocol.FrameHeaderSize + smallMsg.length) + } + + "set compression flag to 1 for large messages (above threshold)" in { + val codec = AdaptiveGzip(1024) + val largeMsg = ByteString.fromArrayUnsafe(new Array[Byte](4096)) + val writer = GrpcProtocolNative.newWriter(codec) + val frame = writer.encodeFrame(GrpcProtocol.DataFrame(largeMsg)) + val chunk = frame.asInstanceOf[pekko.http.scaladsl.model.HttpEntity.Chunk] + val frameData = chunk.data + frameData(0) should be(1.toByte) + } + + "round-trip small message through encode then decode" in { + val codec = AdaptiveGzip(1024) + val writer = GrpcProtocolNative.newWriter(codec) + val reader = GrpcProtocolNative.newReader(Identity) + val originalMsg = ByteString("Hello, adaptive compression!") + val frame = writer.encodeFrame(GrpcProtocol.DataFrame(originalMsg)) + val chunk = frame.asInstanceOf[pekko.http.scaladsl.model.HttpEntity.Chunk] + val decoded = reader.decodeSingleFrame(chunk.data) + decoded should be(originalMsg) + } + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
