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-http.git
The following commit(s) were added to refs/heads/main by this push:
new 7f71b136c perf: avoid copying WebSocket deflate/inflate output (#1235)
7f71b136c is described below
commit 7f71b136c08d76393c97de6e386400e175e6d612
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 11 10:24:07 2026 +0100
perf: avoid copying WebSocket deflate/inflate output (#1235)
* perf: avoid copying WebSocket deflate/inflate output
Motivation:
PerMessageDeflate buffers the output of the Deflater/Inflater in a
ByteArrayOutputStream and then calls ByteString.fromArrayUnsafe on
toByteArray. ByteArrayOutputStream.toByteArray always copies the buffer,
so every compressed or decompressed WebSocket frame pays a full copy of
its payload.
Modification:
Add an internal ByteStringOutputStream that extends
ByteArrayOutputStream and exposes toByteStringUnsafe, wrapping the
internal buffer in a ByteString without copying when most of the buffer
is used, and copying to a right-sized array otherwise so a large buffer
is not retained by a small payload. Use it for the inflate and deflate
paths in PerMessageDeflate, which allocate the stream per call and
discard it immediately afterwards.
Result:
No copy of the payload per WebSocket frame when permessage-deflate is
enabled, and no oversized buffer retained when a frame only fills a
small part of it.
Tests:
- sbt "http-core/testOnly
org.apache.pekko.http.impl.util.ByteStringOutputStreamSpec
org.apache.pekko.http.impl.engine.ws.WebSocketServerSpec" - 51 tests succeeded
- scalafmt --list --mode diff-ref=upstream/main - no files reported
- sbt headerCreateAll - headers added for the new files
References:
None - the ByteStringOutputStream implementation is adapted from Apache
Pekko gRPC (Apache License 2.0),
https://github.com/apache/pekko-grpc/pull/862
* chore: add missing MiMa filters for HttpMessageParser completion handling
Motivation:
#1246 replaced the object-level `CompletionIsMessageStartError` with a
trait-level `completionIsMessageStartError` in the internal
`HttpMessageParser`, but the excludes file it added ends with the
introducing comment and no filter lines, so `validatePullRequest` fails
MiMa on http-core for every PR rebased onto current main.
Modification:
Add the three `HttpMessageParser` filters under the existing comment in
`illegal-request-context.excludes`.
Result:
`sbt "http-core/mimaReportBinaryIssues"` passes again.
---
.../illegal-request-context.excludes | 3 +
.../http/impl/engine/ws/PerMessageDeflate.scala | 10 ++--
.../http/impl/util/ByteStringOutputStream.scala | 51 ++++++++++++++++
.../impl/util/ByteStringOutputStreamSpec.scala | 69 ++++++++++++++++++++++
4 files changed, 128 insertions(+), 5 deletions(-)
diff --git
a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/illegal-request-context.excludes
b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/illegal-request-context.excludes
index a7049a8bf..84b2c9a1a 100644
---
a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/illegal-request-context.excludes
+++
b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/illegal-request-context.excludes
@@ -23,3 +23,6 @@
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.http.impl.e
ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.http.impl.engine.parsing.ParserOutput#MessageStartError.unapply")
# internal API: replaced by an instance level completion handling that can
report the same context
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.http.impl.engine.parsing.HttpMessageParser.CompletionIsMessageStartError")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.impl.engine.parsing.HttpMessageParser.org$apache$pekko$http$impl$engine$parsing$HttpMessageParser$_setter_$completionIsMessageStartError_=")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.impl.engine.parsing.HttpMessageParser.completionIsMessageStartError")
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
index 771da9027..ad5141fab 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
@@ -17,7 +17,6 @@
package org.apache.pekko.http.impl.engine.ws
-import java.io.ByteArrayOutputStream
import java.util.Random
import java.util.zip.Deflater
import java.util.zip.Inflater
@@ -27,6 +26,7 @@ import org.apache.pekko
import pekko.NotUsed
import pekko.annotation.InternalApi
import pekko.http.impl.settings.WebSocketCompressionSettingsImpl
+import pekko.http.impl.util.ByteStringOutputStream
import pekko.http.scaladsl.model.headers.WebSocketExtension
import pekko.stream.scaladsl.BidiFlow
import pekko.stream.scaladsl.Flow
@@ -246,7 +246,7 @@ private[http] object PerMessageDeflate {
try {
val input = if (appendTail) data ++ EmptyStoredBlock else data
inflater.setInput(input.toArrayUnsafe())
- val output = new ByteArrayOutputStream(1024)
+ val output = new ByteStringOutputStream(1024)
var count = inflater.inflate(buffer)
while (count > 0) {
decompressedMessageBytes += count
@@ -255,7 +255,7 @@ private[http] object PerMessageDeflate {
output.write(buffer, 0, count)
count = inflater.inflate(buffer)
}
- ByteString.fromArrayUnsafe(output.toByteArray)
+ output.toByteStringUnsafe
} catch {
case ex: DataFormatException =>
throw new ProtocolException(s"Invalid WebSocket compressed message:
${ex.getMessage}")
@@ -345,13 +345,13 @@ private[http] object PerMessageDeflate {
private def deflate(data: ByteString, removeTail: Boolean): ByteString = {
deflater.setInput(data.toArrayUnsafe())
- val output = new ByteArrayOutputStream(1024)
+ val output = new ByteStringOutputStream(1024)
var count = deflater.deflate(buffer, 0, buffer.length,
Deflater.SYNC_FLUSH)
while (count > 0) {
output.write(buffer, 0, count)
count = deflater.deflate(buffer, 0, buffer.length, Deflater.SYNC_FLUSH)
}
- val bytes = ByteString.fromArrayUnsafe(output.toByteArray)
+ val bytes = output.toByteStringUnsafe
if (removeTail && bytes.endsWith(EmptyStoredBlock))
bytes.dropRight(EmptyStoredBlock.length) else bytes
}
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/util/ByteStringOutputStream.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/util/ByteStringOutputStream.scala
new file mode 100644
index 000000000..acb034ffd
--- /dev/null
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/util/ByteStringOutputStream.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.http.impl.util
+
+import java.io.ByteArrayOutputStream
+
+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 can hand the buffered data over as a [[pekko.util.ByteString]] without
copying it, unlike
+ * `ByteArrayOutputStream.toByteArray` which always creates a copy.
+ *
+ * Derived from the `ByteStringOutputStream` in Apache Pekko gRPC
+ * (https://github.com/apache/pekko-grpc/pull/862).
+ */
+@InternalApi
+private[http] final class ByteStringOutputStream(capacity: Int) extends
ByteArrayOutputStream(capacity) {
+
+ /**
+ * Wraps the bytes written so far in a `ByteString`. The buffer may be
shared with the returned
+ * `ByteString`, so this stream must not be written to, reset or reused
afterwards.
+ */
+ def toByteStringUnsafe: ByteString =
+ if (count < 1) ByteString.empty
+ else if (count > (buf.length >> 1))
+ // Most of the buffer is used — wrap it to avoid a copy
+ ByteString.fromArrayUnsafe(buf, 0, count)
+ else
+ // Small amount of data in a large buffer — copy to right-size so the
rest can be GC'd
+ ByteString.fromArray(buf, 0, count)
+}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/util/ByteStringOutputStreamSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/util/ByteStringOutputStreamSpec.scala
new file mode 100644
index 000000000..a9130162b
--- /dev/null
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/util/ByteStringOutputStreamSpec.scala
@@ -0,0 +1,69 @@
+/*
+ * 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.http.impl.util
+
+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).toByteStringUnsafe 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.toByteStringUnsafe 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.toByteStringUnsafe 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.toByteStringUnsafe 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))
+ // the ByteString is a copy, so it is not affected by later writes to
the stream
+ val result = out.toByteStringUnsafe
+ out.write(Array[Byte](9, 9, 9))
+ result should ===(ByteString(1, 2, 3))
+ }
+
+ "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.toByteStringUnsafe should ===(ByteString(1, 2, 3, 4, 5, 6, 7, 8))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
