From bf6166637edb26d27d8f8d5c16b5fc743f13e4df Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Sun, 13 Sep 2026 16:01:56 +0100 Subject: [PATCH] perf: encode HPACK header blocks from byte arrays into an unsynchronized ByteStringOutputStream 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 --- .../twitter/hpack/HpackEncoderBenchmark.scala | 74 ++++++++++ .../shaded/com/twitter/hpack/Encoder.java | 31 +++-- .../com/twitter/hpack/HuffmanEncoder.java | 57 +++++--- ...-encoder-bytestring-output-stream.excludes | 24 ++++ .../http2/hpack/HeaderCompression.scala | 7 +- .../impl/engine/ws/PerMessageDeflate.scala | 4 +- .../impl/util/ByteStringOutputStream.scala | 90 ++++++++++-- .../pekko/http/impl/util/StringTools.scala | 10 +- .../engine/http2/hpack/HpackDecoderSpec.scala | 7 +- .../util/ByteStringOutputStreamSpec.scala | 49 ++++++- .../http/impl/util/StringToolsSpec.scala | 16 +++ .../com/twitter/hpack/HpackEncoderSpec.scala | 128 ++++++++++++++++++ .../engine/http2/HPackEncodingSupport.scala | 10 +- 13 files changed, 432 insertions(+), 75 deletions(-) create mode 100644 http-bench-jmh/src/main/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderBenchmark.scala create mode 100644 http-core/src/main/mima-filters/2.0.x.backwards.excludes/hpack-encoder-bytestring-output-stream.excludes create mode 100644 http-core/src/test/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderSpec.scala 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 0000000000..0ad3bb9a87 --- /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 c94d6df90d..1bf7062eb1 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 Encoder(int maxHeaderTableSize) { } /** 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 void encodeHeader(OutputStream out, String name, String value, boolean se } /** 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 int getMaxHeaderTableSize() { } /** 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 @@ private static void encodeInteger(OutputStream out, int mask, int n, int i) thro } /** 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 @@ private int getNameIndex(String name) { * 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 7b8771e8e3..413c7affcc 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 + * dst instead of byte by byte to an OutputStream. * - * @param out the output stream for the compressed data - * @throws IOException if an I/O error occurs. In particular, an IOException 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 dst to start writing at + * @param encodedLength the value of {@link #getEncodedLength(byte[])} for data, + * which the caller has typically already computed to decide whether to use Huffman coding and + * to reserve room in dst */ - 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 @@ public void encode(OutputStream out, String string) throws IOException { 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 @@ public void encode(OutputStream out, String string) throws IOException { * @param data the string literal to be Huffman encoded * @return the number of bytes required to Huffman encode data */ - 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 0000000000..d96eda20e5 --- /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 9bce63b8ed..c86ef63674 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 8453fcc2cf..24b4e2c11f 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 acb034ffdc..5d2e77a6a6 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 92823da04e..3eb25e24b8 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 2e0d907113..bde40573b4 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 a9130162bf..f8982e267b 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 ec30db5255..2347668bf4 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 0000000000..8209603d07 --- /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 063e868ff2..4bd5912a65 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() } }