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 cbc42067d perf: encode HPACK header blocks from byte arrays into an
unsynchronized ByteStringOutputStream (#1301)
cbc42067d is described below
commit cbc42067d3372f24f133f6c449b14789a2419227
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Sep 23 08:06:10 2026 +0100
perf: encode HPACK header blocks from byte arrays into an unsynchronized
ByteStringOutputStream (#1301)
Motivation:
HuffmanEncoder read each string literal one char at a time through
String.charAt, twice (once to size it, once to code it), and wrote its
output one byte at a time through OutputStream.write(int). Every write,
including the integer prefixes and bulk literal writes from Encoder,
went through the synchronized ByteArrayOutputStream that
HeaderCompression passes in, and the finished block was copied out with
toByteArray. The per-byte synchronized write dominated the cost of
encoding a header block; the remaining per-header locks were another
fifth of what was left.
Modification:
HuffmanEncoder takes byte[] input, as the upstream twitter/hpack code
did, and codes straight into a reserved range of the output array.
Encoder converts each string literal to octets once (via
StringTools.asciiStringBytes, now ISO-8859-1 and so the exact inverse of
asciiStringFromBytes: a raw literal carrying an octet in 0x80-0xFF sends
that octet where US-ASCII substituted '?', as the Huffman form already
did) and uses them for the length check and both literal forms.
ByteStringOutputStream (#1235) is generalised into a plain OutputStream
with unsynchronized writes, a reserve/array pair for direct fills, and
takeByteString(), which hands the block over without copying when most
of the buffer is used (starting the next block on a fresh array sized to
the last one) or copies out and keeps the array when only a small part
is, as toByteStringUnsafe did. Encoder writes into that stream instead of
any OutputStream; HeaderCompression and PerMessageDeflate use it. One
MiMa excludes file covers the shaded Encoder and HuffmanEncoder
signatures.
Result:
Encoding a typical 10-header browser request is ~2.7x faster on JDK 17
(about 9.6us to 3.5us, see the PR for the JMH table), both HPACK string
literal forms round-trip every octet through the decoder, and the
WebSocket deflate stages no longer lock per write.
Tests:
- sbt "http-core/Test/testOnly
org.apache.pekko.http.impl.util.ByteStringOutputStreamSpec
org.apache.pekko.http.impl.util.StringToolsSpec
org.apache.pekko.http.impl.engine.http2.hpack.HpackDecoderSpec
org.apache.pekko.http.shaded.com.twitter.hpack.*" (Scala 2.13.18 and 3.3.8)
- sbt "http-core/Test/testOnly org.apache.pekko.http.impl.engine.ws.*"
- sbt "http2-tests/Test/testOnly
org.apache.pekko.http.impl.engine.http2.Http2ServerSpec
org.apache.pekko.http.impl.engine.http2.Http2ClientSpec
org.apache.pekko.http.impl.engine.http2.RequestParsingSpec"
- sbt "http-bench-jmh/Jmh/run HpackEncoderBenchmark" on main and on this
branch
- sbt http-core/mimaReportBinaryIssues; sbt headerCheckAll; sbt
javafmtCheckAll (JDK 17)
- scalafmt --mode diff-ref=upstream/main
References:
Refs #1235 - the ByteStringOutputStream this generalises; Refs #1300 -
superseded by this PR
---
.../com/twitter/hpack/HpackEncoderBenchmark.scala | 74 ++++++++++++
.../http/shaded/com/twitter/hpack/Encoder.java | 31 ++---
.../shaded/com/twitter/hpack/HuffmanEncoder.java | 57 +++++----
...hpack-encoder-bytestring-output-stream.excludes | 24 ++++
.../engine/http2/hpack/HeaderCompression.scala | 7 +-
.../http/impl/engine/ws/PerMessageDeflate.scala | 4 +-
.../http/impl/util/ByteStringOutputStream.scala | 90 ++++++++++++---
.../apache/pekko/http/impl/util/StringTools.scala | 10 +-
.../impl/engine/http2/hpack/HpackDecoderSpec.scala | 7 +-
.../impl/util/ByteStringOutputStreamSpec.scala | 49 ++++++--
.../pekko/http/impl/util/StringToolsSpec.scala | 16 +++
.../com/twitter/hpack/HpackEncoderSpec.scala | 128 +++++++++++++++++++++
.../impl/engine/http2/HPackEncodingSupport.scala | 10 +-
13 files changed, 432 insertions(+), 75 deletions(-)
diff --git
a/http-bench-jmh/src/main/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderBenchmark.scala
b/http-bench-jmh/src/main/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderBenchmark.scala
new file mode 100644
index 000000000..0ad3bb9a8
--- /dev/null
+++
b/http-bench-jmh/src/main/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderBenchmark.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.shaded.com.twitter.hpack
+
+import java.util.concurrent.TimeUnit
+
+import org.openjdk.jmh.annotations._
+
+import org.apache.pekko.http.impl.util.ByteStringOutputStream
+
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@BenchmarkMode(Array(Mode.AverageTime))
+@Fork(2)
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 8, time = 2)
+class HpackEncoderBenchmark {
+
+ /** `default` lets the encoder pick Huffman when it is shorter, the other
two force one string literal form. */
+ @Param(Array("default", "huffman", "raw"))
+ var mode = "default"
+
+ // indexing is disabled so that every call encodes the values as string
literals instead of table references
+ var encoder: Encoder = null
+ val out = new ByteStringOutputStream(128)
+
+ // a typical browser request; every value goes through
HuffmanEncoder.getEncodedLength and most through encode
+ val headers: Array[(String, String)] = Array(
+ ":method" -> "GET",
+ ":scheme" -> "https",
+ ":authority" -> "www.example.com",
+ ":path" ->
"/api/v1/users/12345/orders?page=2&sort=created_at&direction=desc",
+ "user-agent" ->
+ "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36
(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+ "accept" ->
"text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
+ "accept-language" -> "en-US,en;q=0.9",
+ "accept-encoding" -> "gzip, deflate, br",
+ "cookie" -> "session=ab12cd34ef56ab12cd34ef56ab12cd34ef56ab12; theme=dark;
consent=1",
+ "cache-control" -> "no-cache")
+
+ @Setup
+ def setup(): Unit =
+ encoder = mode match {
+ case "default" => new Encoder(4096, false, false, false)
+ case "huffman" => new Encoder(4096, false, true, false)
+ case "raw" => new Encoder(4096, false, false, true)
+ }
+
+ @Benchmark
+ def encodeHeaders(): Int = {
+ var i = 0
+ while (i < headers.length) {
+ val (name, value) = headers(i)
+ encoder.encodeHeader(out, name, value, false)
+ i += 1
+ }
+ out.takeByteString().length
+ }
+}
diff --git
a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Encoder.java
b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Encoder.java
index c94d6df90..1bf7062eb 100644
---
a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Encoder.java
+++
b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Encoder.java
@@ -31,9 +31,8 @@
package org.apache.pekko.http.shaded.com.twitter.hpack;
-import java.io.IOException;
-import java.io.OutputStream;
import java.util.Arrays;
+import org.apache.pekko.http.impl.util.ByteStringOutputStream;
import org.apache.pekko.http.impl.util.StringTools;
import org.apache.pekko.http.shaded.com.twitter.hpack.HpackUtil.IndexType;
@@ -75,8 +74,8 @@ public final class Encoder {
}
/** Encode the header field into the header block. */
- public void encodeHeader(OutputStream out, String name, String value,
boolean sensitive)
- throws IOException {
+ public void encodeHeader(
+ ByteStringOutputStream out, String name, String value, boolean
sensitive) {
// If the header value is sensitive then it must never be indexed
if (sensitive) {
@@ -130,7 +129,7 @@ public final class Encoder {
}
/** Set the maximum table size. */
- public void setMaxHeaderTableSize(OutputStream out, int maxHeaderTableSize)
throws IOException {
+ public void setMaxHeaderTableSize(ByteStringOutputStream out, int
maxHeaderTableSize) {
if (maxHeaderTableSize < 0) {
throw new IllegalArgumentException("Illegal Capacity: " +
maxHeaderTableSize);
}
@@ -148,7 +147,7 @@ public final class Encoder {
}
/** Encode integer according to Section 5.1. */
- private static void encodeInteger(OutputStream out, int mask, int n, int i)
throws IOException {
+ private static void encodeInteger(ByteStringOutputStream out, int mask, int
n, int i) {
if (n < 0 || n > 8) {
throw new IllegalArgumentException("N: " + n);
}
@@ -171,23 +170,25 @@ public final class Encoder {
}
/** Encode string literal according to Section 5.2. */
- private void encodeStringLiteral(OutputStream out, String string) throws
IOException {
- int length = string.length();
- int huffmanLength = Huffman.ENCODER.getEncodedLength(string);
+ private void encodeStringLiteral(ByteStringOutputStream out, String string) {
+ // convert once up front: the length computation, the Huffman coder and
the raw literal all work
+ // on the octets
+ byte[] stringBytes = StringTools.asciiStringBytes(string);
+ int length = stringBytes.length;
+ int huffmanLength = Huffman.ENCODER.getEncodedLength(stringBytes);
if ((huffmanLength < length && !forceHuffmanOff) || forceHuffmanOn) {
encodeInteger(out, 0x80, 7, huffmanLength);
- Huffman.ENCODER.encode(out, string);
+ int position = out.reserve(huffmanLength);
+ Huffman.ENCODER.encode(stringBytes, out.array(), position,
huffmanLength);
} else {
- byte[] stringBytes = StringTools.asciiStringBytes(string);
encodeInteger(out, 0x00, 7, length);
- out.write(stringBytes, 0, stringBytes.length);
+ out.write(stringBytes, 0, length);
}
}
/** Encode literal header field according to Section 6.2. */
private void encodeLiteral(
- OutputStream out, String name, String value, IndexType indexType, int
nameIndex)
- throws IOException {
+ ByteStringOutputStream out, String name, String value, IndexType
indexType, int nameIndex) {
int mask;
int prefixBits;
switch (indexType) {
@@ -228,7 +229,7 @@ public final class Encoder {
* Ensure that the dynamic table has enough room to hold 'headerSize' more
bytes. Removes the
* oldest entry from the dynamic table until sufficient space is available.
*/
- private void ensureCapacity(int headerSize) throws IOException {
+ private void ensureCapacity(int headerSize) {
while (size + headerSize > capacity) {
int index = length();
if (index == 0) {
diff --git
a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/HuffmanEncoder.java
b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/HuffmanEncoder.java
index 7b8771e8e..413c7affc 100644
---
a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/HuffmanEncoder.java
+++
b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/HuffmanEncoder.java
@@ -31,9 +31,6 @@
package org.apache.pekko.http.shaded.com.twitter.hpack;
-import java.io.IOException;
-import java.io.OutputStream;
-
final class HuffmanEncoder {
private final int[] codes;
@@ -51,25 +48,37 @@ final class HuffmanEncoder {
}
/**
- * Compresses the input string literal using the Huffman coding.
+ * Compresses the input string literal using the Huffman coding, writing the
result directly into
+ * <code>dst</code> instead of byte by byte to an <code>OutputStream</code>.
*
- * @param out the output stream for the compressed data
- * @throws IOException if an I/O error occurs. In particular, an
<code>IOException</code> may be
- * thrown if the output stream has been closed.
+ * @param data the string literal to be Huffman encoded
+ * @param dst the array to write the Huffman coded string literal to
+ * @param dstOffset the position in <code>dst</code> to start writing at
+ * @param encodedLength the value of {@link #getEncodedLength(byte[])} for
<code>data</code>,
+ * which the caller has typically already computed to decide whether to
use Huffman coding and
+ * to reserve room in <code>dst</code>
*/
- public void encode(OutputStream out, String string) throws IOException {
- if (out == null) {
- throw new NullPointerException("out");
- } else if (string == null) {
- throw new NullPointerException("string");
+ public void encode(byte[] data, byte[] dst, int dstOffset, int
encodedLength) {
+ if (data == null) {
+ throw new NullPointerException("data");
+ } else if (dst == null) {
+ throw new NullPointerException("dst");
+ } else if (dstOffset < 0 || encodedLength < 0 || dstOffset + encodedLength
> dst.length) {
+ throw new IndexOutOfBoundsException(
+ "dstOffset "
+ + dstOffset
+ + ", encodedLength "
+ + encodedLength
+ + ", dst.length "
+ + dst.length);
}
+ int pos = dstOffset;
long current = 0;
int n = 0;
- int len = string.length();
- for (int i = 0; i < len; i++) {
- int b = string.charAt(i) & 0xFF;
+ for (byte value : data) {
+ int b = value & 0xFF;
int code = codes[b];
int nbits = lengths[b];
@@ -79,14 +88,22 @@ final class HuffmanEncoder {
while (n >= 8) {
n -= 8;
- out.write(((int) (current >> n)));
+ dst[pos++] = (byte) (current >> n);
}
}
if (n > 0) {
current <<= (8 - n);
current |= (0xFF >>> n); // this should be EOS symbol
- out.write((int) current);
+ dst[pos++] = (byte) current;
+ }
+
+ if (pos - dstOffset != encodedLength) {
+ throw new IllegalArgumentException(
+ "encodedLength "
+ + encodedLength
+ + " does not match the Huffman encoded length "
+ + (pos - dstOffset));
}
}
@@ -96,13 +113,13 @@ final class HuffmanEncoder {
* @param data the string literal to be Huffman encoded
* @return the number of bytes required to Huffman encode <code>data</code>
*/
- public int getEncodedLength(String data) {
+ public int getEncodedLength(byte[] data) {
if (data == null) {
throw new NullPointerException("data");
}
long len = 0;
- for (int i = 0; i < data.length(); i++) {
- len += lengths[data.charAt(i) & 0xFF];
+ for (byte b : data) {
+ len += lengths[b & 0xFF];
}
return (int) ((len + 7) >> 3);
}
diff --git
a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/hpack-encoder-bytestring-output-stream.excludes
b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/hpack-encoder-bytestring-output-stream.excludes
new file mode 100644
index 000000000..d96eda20e
--- /dev/null
+++
b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/hpack-encoder-bytestring-output-stream.excludes
@@ -0,0 +1,24 @@
+# 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.
+
+# The shaded HPACK Encoder writes into the internal ByteStringOutputStream
instead of any java.io.OutputStream,
+# and the package-private HuffmanEncoder codes byte arrays instead of Strings,
straight into that stream's array
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.Encoder.encodeHeader")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.Encoder.setMaxHeaderTableSize")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.HuffmanEncoder.encode")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.HuffmanEncoder.getEncodedLength")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.HuffmanEncoder.encode")
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderCompression.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderCompression.scala
index 9bce63b8e..c86ef6367 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderCompression.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderCompression.scala
@@ -13,11 +13,11 @@
package org.apache.pekko.http.impl.engine.http2.hpack
-import java.io.ByteArrayOutputStream
import org.apache.pekko
import pekko.annotation.InternalApi
import pekko.http.impl.engine.http2.Http2Protocol.SettingIdentifier
import pekko.http.impl.engine.http2._
+import pekko.http.impl.util.ByteStringOutputStream
import pekko.stream.{ Attributes, FlowShape, Inlet, Outlet }
import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler,
OutHandler, StageLogging }
import pekko.util.ByteString
@@ -44,7 +44,7 @@ private[http2] object HeaderCompression extends
GraphStage[FlowShape[FrameEvent,
private val currentMaxFrameSize = Http2Protocol.InitialMaxFrameSize
val encoder = new
pekko.http.shaded.com.twitter.hpack.Encoder(Http2Protocol.InitialMaxHeaderTableSize)
- val os = new ByteArrayOutputStream(128)
+ val os = new ByteStringOutputStream(128)
def onPull(): Unit = pull(eventsIn)
def onPush(): Unit = grab(eventsIn) match {
@@ -71,8 +71,7 @@ private[http2] object HeaderCompression extends
GraphStage[FlowShape[FrameEvent,
throw new IllegalStateException(
s"Didn't expect key-value-pair [$key] ->
[$value](${value.getClass}) here.")
}
- val result = ByteString.fromArrayUnsafe(os.toByteArray) //
BAOS.toByteArray always creates a copy
- os.reset()
+ val result = os.takeByteString() // hands the array over without
copying and starts a new block
if (result.size <= currentMaxFrameSize)
push(eventsOut, HeadersFrame(streamId, endStream, endHeaders =
true, result, prioInfo))
else {
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 8453fcc2c..24b4e2c11 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
@@ -255,7 +255,7 @@ private[http] object PerMessageDeflate {
output.write(buffer, 0, count)
count = inflater.inflate(buffer)
}
- output.toByteStringUnsafe
+ output.takeByteString()
} catch {
case ex: DataFormatException =>
throw new ProtocolException(s"Invalid WebSocket compressed message:
${ex.getMessage}")
@@ -351,7 +351,7 @@ private[http] object PerMessageDeflate {
output.write(buffer, 0, count)
count = deflater.deflate(buffer, 0, buffer.length, Deflater.SYNC_FLUSH)
}
- val bytes = output.toByteStringUnsafe
+ val bytes = output.takeByteString()
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
index acb034ffd..5d2e77a6a 100644
---
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
@@ -17,7 +17,8 @@
package org.apache.pekko.http.impl.util
-import java.io.ByteArrayOutputStream
+import java.io.OutputStream
+import java.util.Arrays
import org.apache.pekko
import pekko.annotation.InternalApi
@@ -27,25 +28,88 @@ 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.
+ * 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 is a single stage that owns the stream, and the
per-call monitor was a
+ * measurable share of the cost of encoding an HPACK header block.
+ *
+ * After a hand-over the stream is empty again and can be reused for the next
block of data.
+ *
+ * Not thread-safe.
*
* 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) {
+private[http] final class ByteStringOutputStream(initialCapacity: Int) extends
OutputStream {
+ if (initialCapacity < 0) throw new IllegalArgumentException(s"Illegal
initial capacity: $initialCapacity")
+
+ private[this] var buf: Array[Byte] = Array.emptyByteArray
+ private[this] var count: Int = 0
+ // the size of the last block handed over; the array for the next block is
allocated to hold it
+ private[this] 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
+ }
/**
- * 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.
+ * 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 toByteStringUnsafe: ByteString =
+ 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 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)
+ 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/http-core/src/main/scala/org/apache/pekko/http/impl/util/StringTools.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StringTools.scala
index 92823da04..3eb25e24b 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StringTools.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StringTools.scala
@@ -27,9 +27,9 @@ private[http] object StringTools {
// keeps the array as is with a LATIN1 coder, so it is the same single
copy.
new String(bytes, ISO88591)
- def asciiStringBytes(string: String): Array[Byte] = {
- // this is as fast as Unsafe.copyUSAsciiStrToBytes for recent JDK versions
- // and avoids the use of deprecated Unsafe methods
- string.getBytes(java.nio.charset.StandardCharsets.US_ASCII)
- }
+ def asciiStringBytes(string: String): Array[Byte] =
+ // ISO-8859-1 makes this the exact inverse of asciiStringFromBytes: HPACK
string literals are opaque octets and
+ // a character in 0x80-0xFF has to come out as that octet, not as the '?'
that encoding with US-ASCII would
+ // substitute. Since JDK 9 (compact strings) it is a single array copy for
a string with a LATIN1 coder.
+ string.getBytes(ISO88591)
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
index 2e0d90711..bde40573b 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
@@ -17,11 +17,12 @@
package org.apache.pekko.http.impl.engine.http2.hpack
-import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, IOException,
InputStream, SequenceInputStream }
+import java.io.{ ByteArrayInputStream, IOException, InputStream,
SequenceInputStream }
import scala.jdk.CollectionConverters._
import scala.collection.mutable.ListBuffer
+import org.apache.pekko.http.impl.util.ByteStringOutputStream
import org.apache.pekko.http.shaded.com.twitter.hpack.{ Decoder, Encoder,
HeaderListener }
import org.scalatest.matchers.should.Matchers
@@ -57,10 +58,10 @@ class HpackDecoderSpec extends AnyWordSpec with Matchers {
bytes.grouped(chunkSize).map(chunk => new ByteArrayInputStream(chunk):
InputStream).asJavaEnumeration)
private def encode(headers: (String, String)*): Array[Byte] = {
- val out = new ByteArrayOutputStream
+ val out = new ByteStringOutputStream(128)
val encoder = new Encoder(maxHeaderTableSize)
headers.foreach { case (name, value) => encoder.encodeHeader(out, name,
value, false) }
- out.toByteArray
+ out.takeByteString().toArray
}
private def decode(in: InputStream): Seq[(String, String)] = {
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
index a9130162b..f8982e267 100644
---
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
@@ -26,36 +26,53 @@ class ByteStringOutputStreamSpec extends AnyWordSpec with
Matchers {
"ByteStringOutputStream" must {
"return an empty ByteString when nothing was written" in {
- new ByteStringOutputStream(16).toByteStringUnsafe should
===(ByteString.empty)
+ 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.toByteStringUnsafe should ===(ByteString(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.toByteStringUnsafe should ===(ByteString(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.toByteStringUnsafe should ===(ByteString(1, 2, 3, 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))
- // the ByteString is a copy, so it is not affected by later writes to
the stream
- val result = out.toByteStringUnsafe
+ 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 {
@@ -63,7 +80,25 @@ class ByteStringOutputStreamSpec extends AnyWordSpec with
Matchers {
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))
+ 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)
}
}
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/util/StringToolsSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/util/StringToolsSpec.scala
index ec30db525..2347668bf 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/util/StringToolsSpec.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/util/StringToolsSpec.scala
@@ -50,5 +50,21 @@ class StringToolsSpec extends AnyWordSpec with Matchers {
"encode an ASCII string to its US-ASCII bytes" in {
StringTools.asciiStringBytes("abc") shouldEqual
"abc".getBytes(StandardCharsets.US_ASCII)
}
+
+ "be the inverse of asciiStringFromBytes for every octet" in {
+ // HPACK string literals are opaque octets, so a char in 0x80-0xFF must
map back to that octet rather than
+ // to the '?' that encoding with US-ASCII would substitute
+ val allOctets = Array.tabulate(256)(_.toByte)
+
StringTools.asciiStringBytes(StringTools.asciiStringFromBytes(allOctets))
shouldEqual allOctets
+ }
+
+ "encode a character above 0xFF, which no octet can represent, as a single
'?'" in {
+ // U+0100 forces the UTF-16 coder on JDK 9+, so this covers the
non-arraycopy path too
+ StringTools.asciiStringBytes("aĀb") shouldEqual Array[Byte]('a', '?',
'b')
+ }
+
+ "encode the empty string to an empty array" in {
+ StringTools.asciiStringBytes("") shouldEqual Array.emptyByteArray
+ }
}
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderSpec.scala
new file mode 100644
index 000000000..8209603d0
--- /dev/null
+++
b/http-core/src/test/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderSpec.scala
@@ -0,0 +1,128 @@
+/*
+ * 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.shaded.com.twitter.hpack
+
+import java.io.ByteArrayInputStream
+
+import scala.collection.immutable.VectorBuilder
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import org.apache.pekko.http.impl.util.{ ByteStringOutputStream, StringTools }
+
+class HpackEncoderSpec extends AnyWordSpec with Matchers {
+
+ private def hex(bytes: Array[Byte]): String = bytes.map(b =>
f"$b%02x").mkString
+
+ private def huffmanEncode(bytes: Array[Byte]): Array[Byte] = {
+ val encodedLength = Huffman.ENCODER.getEncodedLength(bytes)
+ val dst = new Array[Byte](encodedLength)
+ Huffman.ENCODER.encode(bytes, dst, 0, encodedLength)
+ dst
+ }
+
+ private def huffmanEncode(literal: String): Array[Byte] =
huffmanEncode(StringTools.asciiStringBytes(literal))
+
+ // indexing off so that the header value is always sent as a string literal,
never as a table reference
+ private def encodeHeader(name: String, value: String, forceHuffmanOn:
Boolean, forceHuffmanOff: Boolean)
+ : Array[Byte] = {
+ val encoder = new Encoder(4096, false, forceHuffmanOn, forceHuffmanOff)
+ val out = new ByteStringOutputStream(16)
+ encoder.encodeHeader(out, name, value, false)
+ out.takeByteString().toArray
+ }
+
+ private def decodeHeaders(bytes: Array[Byte]): Seq[(String, String)] = {
+ val decoder = new Decoder(8192, 4096)
+ val headers = new VectorBuilder[(String, String)]()
+ decoder.decode(new ByteArrayInputStream(bytes),
+ (name: String, value: String, parsedValue: AnyRef, _: Boolean) => {
+ headers += name -> value
+ parsedValue
+ })
+ headers.result()
+ }
+
+ "HuffmanEncoder" should {
+ // the string literals from RFC 7541 Appendix C.4 with their expected
Huffman codes
+ val rfc7541Examples = Seq(
+ "www.example.com" -> "f1e3c2e5f23a6ba0ab90f4ff",
+ "no-cache" -> "a8eb10649cbf",
+ "custom-key" -> "25a849e95ba97d7f",
+ "custom-value" -> "25a849e95bb8e8b4bf")
+
+ "produce the Huffman codes from RFC 7541 Appendix C.4" in {
+ rfc7541Examples.foreach {
+ case (literal, expected) =>
+ hex(huffmanEncode(literal)) shouldEqual expected
+ }
+ }
+
+ "report the encoded length that encode actually produces" in {
+ rfc7541Examples.foreach {
+ case (literal, expected) =>
+
Huffman.ENCODER.getEncodedLength(StringTools.asciiStringBytes(literal))
shouldEqual expected.length / 2
+ }
+ }
+
+ "encode the empty literal to no bytes" in {
+ huffmanEncode("") shouldEqual Array.emptyByteArray
+ Huffman.ENCODER.getEncodedLength(Array.emptyByteArray) shouldEqual 0
+ }
+
+ "encode every octet with the code for that octet" in {
+ // a byte above 0x7F must be looked up as an unsigned symbol, not as a
negative array index
+ val allOctets = Array.tabulate(256)(_.toByte)
+ Huffman.DECODER.decode(huffmanEncode(allOctets)) shouldEqual allOctets
+ }
+
+ "write at the given offset and reject an encoded length that does not
match the input" in {
+ val bytes = StringTools.asciiStringBytes("no-cache")
+ val dst = new Array[Byte](10)
+ Huffman.ENCODER.encode(bytes, dst, 2, 6)
+ hex(dst) shouldEqual "0000a8eb10649cbf0000"
+ an[IllegalArgumentException] should be thrownBy
Huffman.ENCODER.encode(bytes, new Array[Byte](10), 0, 7)
+ an[IndexOutOfBoundsException] should be thrownBy
Huffman.ENCODER.encode(bytes, new Array[Byte](10), 5, 6)
+ }
+ }
+
+ "Encoder" should {
+ "encode the value from RFC 7541 Appendix C.4.1 with Huffman coding when it
is shorter" in {
+ // 0x01: literal without indexing, name index 1 (:authority); 0x8c:
Huffman coded, 12 octets
+ hex(encodeHeader(":authority", "www.example.com", forceHuffmanOn =
false, forceHuffmanOff = false)) shouldEqual
+ "018cf1e3c2e5f23a6ba0ab90f4ff"
+ }
+
+ "encode the header from RFC 7541 Appendix C.2.1 as a raw literal when
Huffman coding is forced off" in {
+ // 0x00: literal without indexing, new name; 0x0a/0x0d: raw string
literals of 10 and 13 octets
+ hex(encodeHeader("custom-key", "custom-header", forceHuffmanOn = false,
forceHuffmanOff = true)) shouldEqual
+ "000a637573746f6d2d6b65790d637573746f6d2d686561646572"
+ }
+
+ "round-trip a value carrying opaque octets above 0x7F in both string
literal forms" in {
+ // HPACK string literals are opaque octets: a decoder maps each byte to
the char of the same value, so an
+ // encoder has to map each such char back to that byte, in the raw form
as well as the Huffman form
+ val value = new String(Array.tabulate(256)(_.toChar))
+ Seq(true, false).foreach { huffman =>
+ val bytes = encodeHeader("x-opaque", value, forceHuffmanOn = huffman,
forceHuffmanOff = !huffman)
+ decodeHeaders(bytes) shouldEqual Seq("x-opaque" -> value)
+ }
+ }
+ }
+}
diff --git
a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/HPackEncodingSupport.scala
b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/HPackEncodingSupport.scala
index 063e868ff..4bd5912a6 100644
---
a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/HPackEncodingSupport.scala
+++
b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/HPackEncodingSupport.scala
@@ -13,12 +13,10 @@
package org.apache.pekko.http.impl.engine.http2
-import java.io.ByteArrayOutputStream
-
import org.apache.pekko
-import pekko.http.impl.util.StringRendering
import pekko.http.scaladsl.model.{ HttpHeader, HttpRequest, HttpResponse }
import pekko.http.scaladsl.model.headers.RawHeader
+import pekko.http.impl.util.{ ByteStringOutputStream, StringRendering }
import pekko.http.shaded.com.twitter.hpack.Encoder
import pekko.util.ByteString
@@ -62,12 +60,12 @@ trait HPackEncodingSupport {
headers.map(h => h.lowercaseName -> h.value)
def encodeHeaderPairs(headerPairs: Seq[(String, String)]): ByteString = {
- val bos = new ByteArrayOutputStream()
+ val out = new ByteStringOutputStream(128)
- def encode(name: String, value: String): Unit = encoder.encodeHeader(bos,
name, value, false)
+ def encode(name: String, value: String): Unit = encoder.encodeHeader(out,
name, value, false)
headerPairs.foreach((encode _).tupled)
- ByteString(bos.toByteArray)
+ out.takeByteString()
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
