This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-grpc.git
The following commit(s) were added to refs/heads/main by this push:
new a670bbce Avoid buffer copies with a non-synchronized
ByteStringOutputStream (#907)
a670bbce is described below
commit a670bbcec5b3f00d6ec01be1d34edeae0bba8dbd
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Oct 13 10:34:26 2026 +0100
Avoid buffer copies with a non-synchronized ByteStringOutputStream (#907)
* Avoid buffer copy on Gzip compress/uncompress.
* Use the non-synchronized ByteStringOutputStream from Pekko HTTP.
Copied from apache/pekko-http http-core ByteStringOutputStream, which hands
the
buffered data over as a ByteString without copying and avoids the
synchronized
writes of ByteArrayOutputStream.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* Avoid private[this], which Scala 3.9 warns about.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* Add a benchmark for gzip and InputStream buffering.
Motivation:
The buffers used by Gzip.compress, Gzip.uncompress and
ByteStringUtils.fromInputStream had no benchmark, so changes to them
could not be measured.
Modification:
Add GzipBenchmark, covering compress, uncompress and fromInputStream
for 1 KiB, 64 KiB and 1 MiB payloads.
Result:
The ByteStringOutputStream change can be compared with main.
Tests:
- sbt benchmarks/Jmh/compile: passed
- sbt "benchmarks/Jmh/run -prof gc org.apache.pekko.grpc.GzipBenchmark":
run on main and this branch
References:
Refs #905
* Avoid deprecated = _ initializers in GzipBenchmark.
Scala 3.9 deprecates them, which fails the next build under -Werror.
---------
Co-authored-by: Tim Whittington <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-authored-by: PJ Fanning <[email protected]>
---
.../org/apache/pekko/grpc/GzipBenchmark.scala | 63 ++++++++++++
.../grpc/internal/ByteStringOutputStream.scala | 113 +++++++++++++++++++++
.../pekko/grpc/internal/ByteStringUtils.scala | 34 ++++---
.../org/apache/pekko/grpc/internal/Gzip.scala | 30 ++----
.../grpc/internal/ByteStringOutputStreamSpec.scala | 104 +++++++++++++++++++
5 files changed, 307 insertions(+), 37 deletions(-)
diff --git
a/benchmarks/src/main/scala/org/apache/pekko/grpc/GzipBenchmark.scala
b/benchmarks/src/main/scala/org/apache/pekko/grpc/GzipBenchmark.scala
new file mode 100644
index 00000000..07c380d6
--- /dev/null
+++ b/benchmarks/src/main/scala/org/apache/pekko/grpc/GzipBenchmark.scala
@@ -0,0 +1,63 @@
+/*
+ * 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
+
+import java.io.ByteArrayInputStream
+
+import scala.util.Random
+
+import org.apache.pekko
+import pekko.grpc.internal.{ ByteStringUtils, Gzip }
+import pekko.util.ByteString
+import org.openjdk.jmh.annotations._
+
+// Microbenchmarks for the buffers used by gzip compression and by reading a
message from an InputStream.
+class GzipBenchmark extends CommonBenchmark {
+ @Param(Array("1024", "65536", "1048576"))
+ var size: Int = 0
+
+ private var uncompressed: ByteString = ByteString.empty
+ private var compressed: ByteString = ByteString.empty
+ private var uncompressedArray: Array[Byte] = Array.emptyByteArray
+
+ @Setup
+ def setup(): Unit = {
+ // Compressible text, closer to a typical message than random bytes
+ val random = new Random(42)
+ val words = Vector.fill(256)(random.alphanumeric.take(3 +
random.nextInt(8)).mkString)
+ val builder = new StringBuilder(size + 16)
+ while (builder.length < size)
builder.append(words(random.nextInt(words.size))).append(' ')
+ uncompressedArray = builder.toString.take(size).getBytes("UTF-8")
+ uncompressed = ByteString(uncompressedArray)
+ compressed = Gzip.compress(uncompressed)
+ }
+
+ @Benchmark
+ def compress(): ByteString =
+ Gzip.compress(uncompressed)
+
+ @Benchmark
+ def uncompress(): ByteString =
+ Gzip.uncompress(compressed, Int.MaxValue)
+
+ // A plain ByteArrayInputStream is not a KnownLength stream, so messages
larger than the 32 KiB read buffer
+ // take the buffering path.
+ @Benchmark
+ def fromInputStream(): ByteString =
+ ByteStringUtils.fromInputStream(new
ByteArrayInputStream(uncompressedArray))
+}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/ByteStringOutputStream.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/ByteStringOutputStream.scala
new file mode 100644
index 00000000..4d64c777
--- /dev/null
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/ByteStringOutputStream.scala
@@ -0,0 +1,113 @@
+/*
+ * 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.OutputStream
+import java.util.Arrays
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.util.ByteString
+
+/**
+ * INTERNAL API
+ *
+ * An [[java.io.OutputStream]] that buffers into a byte array like
[[java.io.ByteArrayOutputStream]] but
+ * that hands the buffered data over as a [[pekko.util.ByteString]] without
copying it, unlike
+ * `ByteArrayOutputStream.toByteArray` which always creates a copy, and whose
writes are not
+ * `synchronized`: every user owns the stream it writes to, so the per-call
monitor is pure overhead.
+ *
+ * After a hand-over the stream is empty again and can be reused for the next
block of data.
+ *
+ * Not thread-safe.
+ *
+ * Copied from the `ByteStringOutputStream` in Apache Pekko HTTP.
+ */
+@InternalApi
+private[grpc] final class ByteStringOutputStream(initialCapacity: Int) extends
OutputStream {
+ if (initialCapacity < 0) throw new IllegalArgumentException(s"Illegal
initial capacity: $initialCapacity")
+
+ private var buf: Array[Byte] = Array.emptyByteArray
+ private var count: Int = 0
+ // the size of the last block handed over; the array for the next block is
allocated to hold it
+ private var lastCount: Int = 0
+
+ /** The number of bytes written since the last hand-over. */
+ def size: Int = count
+
+ override def write(b: Int): Unit = {
+ ensureCapacity(1)
+ buf(count) = b.toByte
+ count += 1
+ }
+
+ override def write(bytes: Array[Byte], offset: Int, length: Int): Unit = {
+ ensureCapacity(length)
+ System.arraycopy(bytes, offset, buf, count, length)
+ count += length
+ }
+
+ /**
+ * Reserves `length` bytes at the end of the buffered data for the caller to
fill in directly.
+ *
+ * @return the position in [[array]] at which the reserved bytes start
+ */
+ def reserve(length: Int): Int = {
+ ensureCapacity(length)
+ val position = count
+ count += length
+ position
+ }
+
+ /**
+ * The array backing the buffered data. Only valid until the next write,
reserve or hand-over, any of
+ * which may replace it.
+ */
+ def array: Array[Byte] = buf
+
+ /**
+ * Hands the buffered data over as a `ByteString` and leaves the stream
empty for the next block.
+ *
+ * When most of the buffer is used the `ByteString` wraps the array without
copying and the stream
+ * starts a fresh array for the next block; when only a small part is used
the data is copied so the
+ * `ByteString` does not retain a large, mostly unused array, and the stream
keeps its array.
+ */
+ def takeByteString(): ByteString =
+ if (count < 1) ByteString.empty
+ else {
+ val result =
+ if (count > (buf.length >> 1)) {
+ val wrapped = ByteString.fromArrayUnsafe(buf, 0, count)
+ lastCount = count
+ buf = Array.emptyByteArray
+ wrapped
+ } else ByteString.fromArray(buf, 0, count)
+ count = 0
+ result
+ }
+
+ private def ensureCapacity(additional: Int): Unit = {
+ val required = count + additional
+ if (required > buf.length) {
+ val capacity =
+ if (buf.length == 0) math.max(math.max(initialCapacity, lastCount),
required)
+ else math.max(buf.length << 1, required)
+ buf = Arrays.copyOf(buf, capacity)
+ }
+ }
+}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/ByteStringUtils.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/ByteStringUtils.scala
index f970c862..7ffcf2ec 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/ByteStringUtils.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/ByteStringUtils.scala
@@ -18,7 +18,7 @@ import pekko.annotation.InternalApi
import pekko.util.ByteString
import io.grpc.KnownLength
-import java.io.{ ByteArrayOutputStream, InputStream }
+import java.io.InputStream
@InternalApi
private[grpc] object ByteStringUtils {
@@ -34,26 +34,30 @@ private[grpc] object ByteStringUtils {
val initialBytes = stream.read(buffer, 0, buffer.length)
val nextByte = if (initialBytes < 0) -1 else stream.read() // Test for EOF
- if (nextByte == -1) {
- if (initialBytes < 1) ByteString.empty // EOF immediately
- else if (initialBytes > (buffer.length >> 1))
- // Most of the buffer is used — reuse it to avoid a copy
- ByteString.fromArrayUnsafe(buffer, 0, initialBytes)
- else
- // Small read from a large buffer — copy to right-size so the rest can
be GC'd
- ByteString.fromArray(buffer, 0, initialBytes)
- } else {
- val baos = new ByteArrayOutputStream(buffer.length * 2) // To avoid
immediate resize
- baos.write(buffer, 0, initialBytes)
- baos.write(nextByte)
+ if (nextByte == -1)
+ toByteStringUnsafe(buffer, initialBytes)
+ else {
+ val bsos = new ByteStringOutputStream(buffer.length * 2) // To avoid
immediate resize
+ bsos.write(buffer, 0, initialBytes)
+ bsos.write(nextByte)
var bytesRead = stream.read(buffer)
while (bytesRead >= 0) {
- baos.write(buffer, 0, bytesRead)
+ bsos.write(buffer, 0, bytesRead)
bytesRead = stream.read(buffer)
}
- ByteString.fromArrayUnsafe(baos.toByteArray)
+ bsos.takeByteString()
}
}
+
+ def toByteStringUnsafe(buf: Array[Byte], count: Int): ByteString =
+ if (count < 1)
+ ByteString.empty
+ else if (count > (buf.length >> 1))
+ // Most of the buffer is used — reuse it to avoid a copy
+ ByteString.fromArrayUnsafe(buf, 0, count)
+ else
+ // Small read from a large buffer — copy to right-size so the rest can
be GC'd
+ ByteString.fromArray(buf, 0, count)
}
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Gzip.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Gzip.scala
index 21d3687b..643b986b 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Gzip.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Gzip.scala
@@ -13,7 +13,6 @@
package org.apache.pekko.grpc.internal
-import java.io.ByteArrayOutputStream
import java.util.zip.{ GZIPInputStream, GZIPOutputStream }
import org.apache.pekko.util.ByteString
@@ -25,28 +24,15 @@ object Gzip extends Codec {
override val name: String = "gzip"
override def compress(uncompressed: ByteString): ByteString = {
- val baos = new ByteArrayOutputStream(uncompressed.size)
- val gzos = new GZIPOutputStream(baos)
+ val bsos = new ByteStringOutputStream(uncompressed.size)
+ val gzos = new GZIPOutputStream(bsos)
try gzos.write(uncompressed.toArrayUnsafe())
finally gzos.close()
- ByteString.fromArrayUnsafe(baos.toByteArray)
+ bsos.takeByteString()
}
@deprecated("Use uncompress(bytes, maxDecompressedSize), which bounds the
decompressed size", "2.0.0")
- override def uncompress(compressed: ByteString): ByteString = {
- val gzis = new GZIPInputStream(compressed.asInputStream)
-
- val baos = new ByteArrayOutputStream(compressed.size)
- val buffer = new Array[Byte](32 * 1024)
- try {
- var read = gzis.read(buffer)
- while (read != -1) {
- baos.write(buffer, 0, read)
- read = gzis.read(buffer)
- }
- } finally gzis.close()
- ByteString.fromArrayUnsafe(baos.toByteArray)
- }
+ override def uncompress(compressed: ByteString): ByteString =
uncompress(compressed, Int.MaxValue)
/**
* Decompress with a maximum decompressed size limit.
@@ -57,10 +43,10 @@ object Gzip extends Codec {
val limit = maxDecompressedSize.toLong
// clamp: maxDecompressedSize is validated to be positive by the settings
classes, but this
// method is also reachable with a hand-constructed limit, and a negative
initial size would
- // make ByteArrayOutputStream throw IllegalArgumentException rather than a
gRPC status.
+ // make ByteStringOutputStream throw IllegalArgumentException rather than
a gRPC status.
val initialSize = Math.max(0L, Math.min(compressed.size.toLong,
limit)).toInt
val gzis = new GZIPInputStream(compressed.asInputStream)
- val baos = new ByteArrayOutputStream(initialSize)
+ val bsos = new ByteStringOutputStream(initialSize)
val buffer = new Array[Byte](32 * 1024)
// Long, so that a limit close to Int.MaxValue cannot be passed by an
overflowing counter
var totalBytes = 0L
@@ -72,11 +58,11 @@ object Gzip extends Codec {
throw new StatusException(
Status.RESOURCE_EXHAUSTED.withDescription(
s"Decompressed message size exceeds maximum allowed
$maxDecompressedSize bytes"))
- baos.write(buffer, 0, read)
+ bsos.write(buffer, 0, read)
read = gzis.read(buffer)
}
} finally gzis.close()
- ByteString.fromArrayUnsafe(baos.toByteArray)
+ bsos.takeByteString()
}
@deprecated(
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/internal/ByteStringOutputStreamSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/internal/ByteStringOutputStreamSpec.scala
new file mode 100644
index 00000000..6ce3e126
--- /dev/null
+++
b/runtime/src/test/scala/org/apache/pekko/grpc/internal/ByteStringOutputStreamSpec.scala
@@ -0,0 +1,104 @@
+/*
+ * 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.util.ByteString
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class ByteStringOutputStreamSpec extends AnyWordSpec with Matchers {
+
+ "ByteStringOutputStream" must {
+
+ "return an empty ByteString when nothing was written" in {
+ new ByteStringOutputStream(16).takeByteString() should
===(ByteString.empty)
+ }
+
+ "return the bytes written when the buffer is exactly filled" in {
+ val out = new ByteStringOutputStream(4)
+ out.write(Array[Byte](1, 2, 3, 4))
+ out.takeByteString() should ===(ByteString(1, 2, 3, 4))
+ }
+
+ "return the bytes written when the buffer was grown" in {
+ val out = new ByteStringOutputStream(2)
+ val data = Array.tabulate[Byte](1000)(i => i.toByte)
+ out.write(data)
+ out.takeByteString() should ===(ByteString(data))
+ }
+
+ "return the bytes written when only a small part of the buffer is used" in
{
+ val out = new ByteStringOutputStream(1024)
+ out.write(Array[Byte](1, 2, 3))
+ out.write(4)
+ out.takeByteString() should ===(ByteString(1, 2, 3, 4))
+ }
+
+ "not retain the buffer when only a small part of it is used" in {
+ val out = new ByteStringOutputStream(1024)
+ out.write(Array[Byte](1, 2, 3))
+ val array = out.array
+ // the ByteString is a copy, so it is not affected by later writes to
the stream, which keeps its array
+ val result = out.takeByteString()
+ out.write(Array[Byte](9, 9, 9))
+ result should ===(ByteString(1, 2, 3))
+ (out.array eq array) shouldBe true
+ }
+
+ "hand over a mostly used buffer without copying and start the next block
on a fresh array" in {
+ val out = new ByteStringOutputStream(4)
+ out.write(Array[Byte](1, 2, 3))
+ val array = out.array
+ val first = out.takeByteString()
+ first should ===(ByteString(1, 2, 3))
+ out.size shouldBe 0
+
+ // writes to the next block must not show up in the ByteString that was
handed out
+ out.write(Array[Byte](9, 9, 9, 9), 0, 4)
+ (out.array eq array) shouldBe false
+ first should ===(ByteString(1, 2, 3))
+ out.takeByteString() should ===(ByteString(9, 9, 9, 9))
+ }
+
+ "write single bytes and byte ranges" in {
+ val out = new ByteStringOutputStream(8)
+ out.write(1)
+ out.write(Array[Byte](0, 2, 3, 0), 1, 2)
+ out.write(Array[Byte](4, 5, 6, 7, 8))
+ out.takeByteString() should ===(ByteString(1, 2, 3, 4, 5, 6, 7, 8))
+ }
+
+ "reserve a range that the caller fills in through the backing array" in {
+ val out = new ByteStringOutputStream(2)
+ out.write(7)
+ val position = out.reserve(3)
+ position shouldBe 1
+ out.size shouldBe 4
+ val array = out.array
+ array(position) = 1
+ array(position + 1) = 2
+ array(position + 2) = 3
+ out.write(8)
+ out.takeByteString() should ===(ByteString(7, 1, 2, 3, 8))
+ }
+
+ "reject a negative initial capacity" in {
+ an[IllegalArgumentException] should be thrownBy new
ByteStringOutputStream(-1)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
